summaryrefslogtreecommitdiff
path: root/tests/unit/ec2
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unit/ec2')
-rw-r--r--tests/unit/ec2/autoscale/test_group.py256
-rw-r--r--tests/unit/ec2/elb/test_loadbalancer.py97
-rw-r--r--tests/unit/ec2/test_address.py16
-rw-r--r--tests/unit/ec2/test_blockdevicemapping.py54
-rw-r--r--tests/unit/ec2/test_connection.py657
-rw-r--r--tests/unit/ec2/test_instance.py2
-rw-r--r--tests/unit/ec2/test_networkinterface.py148
-rw-r--r--tests/unit/ec2/test_securitygroup.py212
-rw-r--r--tests/unit/ec2/test_volume.py37
9 files changed, 1422 insertions, 57 deletions
diff --git a/tests/unit/ec2/autoscale/test_group.py b/tests/unit/ec2/autoscale/test_group.py
index 28941545..b3a5594e 100644
--- a/tests/unit/ec2/autoscale/test_group.py
+++ b/tests/unit/ec2/autoscale/test_group.py
@@ -28,7 +28,12 @@ from tests.unit import AWSMockServiceTestCase
from boto.ec2.autoscale import AutoScaleConnection
from boto.ec2.autoscale.group import AutoScalingGroup
+from boto.ec2.autoscale.policy import ScalingPolicy
+from boto.ec2.autoscale.tag import Tag
+from boto.ec2.blockdevicemapping import EBSBlockDeviceType, BlockDeviceMapping
+
+from boto.ec2.autoscale import launchconfig
class TestAutoScaleGroup(AWSMockServiceTestCase):
connection_class = AutoScaleConnection
@@ -195,6 +200,257 @@ class TestDescribeTerminationPolicies(AWSMockServiceTestCase):
['ClosestToNextInstanceHour', 'Default',
'NewestInstance', 'OldestInstance', 'OldestLaunchConfiguration'])
+class TestLaunchConfiguration(AWSMockServiceTestCase):
+ connection_class = AutoScaleConnection
+
+ def default_body(self):
+ # This is a dummy response
+ return """
+ <DescribeLaunchConfigurationsResponse>
+ </DescribeLaunchConfigurationsResponse>
+ """
+
+ def test_launch_config(self):
+ # This unit test is based on #753 and #1343
+ self.set_http_response(status_code=200)
+ dev_sdf = EBSBlockDeviceType(snapshot_id='snap-12345')
+ dev_sdg = EBSBlockDeviceType(snapshot_id='snap-12346')
+
+ bdm = BlockDeviceMapping()
+ bdm['/dev/sdf'] = dev_sdf
+ bdm['/dev/sdg'] = dev_sdg
+
+ lc = launchconfig.LaunchConfiguration(
+ connection=self.service_connection,
+ name='launch_config',
+ image_id='123456',
+ instance_type = 'm1.large',
+ security_groups = ['group1', 'group2'],
+ spot_price='price',
+ block_device_mappings = [bdm]
+ )
+
+ response = self.service_connection.create_launch_configuration(lc)
+
+ self.assert_request_parameters({
+ 'Action': 'CreateLaunchConfiguration',
+ 'BlockDeviceMappings.member.1.DeviceName': '/dev/sdf',
+ 'BlockDeviceMappings.member.1.Ebs.DeleteOnTermination': 'false',
+ 'BlockDeviceMappings.member.1.Ebs.SnapshotId': 'snap-12345',
+ 'BlockDeviceMappings.member.2.DeviceName': '/dev/sdg',
+ 'BlockDeviceMappings.member.2.Ebs.DeleteOnTermination': 'false',
+ 'BlockDeviceMappings.member.2.Ebs.SnapshotId': 'snap-12346',
+ 'EbsOptimized': 'false',
+ 'LaunchConfigurationName': 'launch_config',
+ 'ImageId': '123456',
+ 'InstanceMonitoring.Enabled': 'false',
+ 'InstanceType': 'm1.large',
+ 'SecurityGroups.member.1': 'group1',
+ 'SecurityGroups.member.2': 'group2',
+ 'SpotPrice': 'price',
+ }, ignore_params_values=['Version'])
+
+
+class TestCreateAutoScalePolicy(AWSMockServiceTestCase):
+ connection_class = AutoScaleConnection
+
+ def setUp(self):
+ super(TestCreateAutoScalePolicy, self).setUp()
+
+ def default_body(self):
+ return """
+ <PutScalingPolicyResponse xmlns="http://autoscaling.amazonaws.com\
+ /doc/2011-01-01/">
+ <PutScalingPolicyResult>
+ <PolicyARN>arn:aws:autoscaling:us-east-1:803981987763:scaling\
+ Policy:b0dcf5e8
+ -02e6-4e31-9719-0675d0dc31ae:autoScalingGroupName/my-test-asg:\
+ policyName/my-scal
+ eout-policy</PolicyARN>
+ </PutScalingPolicyResult>
+ <ResponseMetadata>
+ <RequestId>3cfc6fef-c08b-11e2-a697-2922EXAMPLE</RequestId>
+ </ResponseMetadata>
+ </PutScalingPolicyResponse>
+ """
+
+ def test_scaling_policy_with_min_adjustment_step(self):
+ self.set_http_response(status_code=200)
+
+ policy = ScalingPolicy(
+ name='foo', as_name='bar',
+ adjustment_type='PercentChangeInCapacity', scaling_adjustment=50,
+ min_adjustment_step=30)
+ self.service_connection.create_scaling_policy(policy)
+
+ self.assert_request_parameters({
+ 'Action': 'PutScalingPolicy',
+ 'PolicyName': 'foo',
+ 'AutoScalingGroupName': 'bar',
+ 'AdjustmentType': 'PercentChangeInCapacity',
+ 'ScalingAdjustment': 50,
+ 'MinAdjustmentStep': 30
+ }, ignore_params_values=['Version'])
+
+ def test_scaling_policy_with_wrong_adjustment_type(self):
+ self.set_http_response(status_code=200)
+
+ policy = ScalingPolicy(
+ name='foo', as_name='bar',
+ adjustment_type='ChangeInCapacity', scaling_adjustment=50,
+ min_adjustment_step=30)
+ self.service_connection.create_scaling_policy(policy)
+
+ self.assert_request_parameters({
+ 'Action': 'PutScalingPolicy',
+ 'PolicyName': 'foo',
+ 'AutoScalingGroupName': 'bar',
+ 'AdjustmentType': 'ChangeInCapacity',
+ 'ScalingAdjustment': 50
+ }, ignore_params_values=['Version'])
+
+ def test_scaling_policy_without_min_adjustment_step(self):
+ self.set_http_response(status_code=200)
+
+ policy = ScalingPolicy(
+ name='foo', as_name='bar',
+ adjustment_type='PercentChangeInCapacity', scaling_adjustment=50)
+ self.service_connection.create_scaling_policy(policy)
+
+ self.assert_request_parameters({
+ 'Action': 'PutScalingPolicy',
+ 'PolicyName': 'foo',
+ 'AutoScalingGroupName': 'bar',
+ 'AdjustmentType': 'PercentChangeInCapacity',
+ 'ScalingAdjustment': 50
+ }, ignore_params_values=['Version'])
+
+
+class TestPutNotificationConfiguration(AWSMockServiceTestCase):
+ connection_class = AutoScaleConnection
+
+ def setUp(self):
+ super(TestPutNotificationConfiguration, self).setUp()
+
+ def default_body(self):
+ return """
+ <PutNotificationConfigurationResponse>
+ <ResponseMetadata>
+ <RequestId>requestid</RequestId>
+ </ResponseMetadata>
+ </PutNotificationConfigurationResponse>
+ """
+
+ def test_autoscaling_group_put_notification_configuration(self):
+ self.set_http_response(status_code=200)
+ autoscale = AutoScalingGroup(
+ name='ana', launch_config='lauch_config',
+ min_size=1, max_size=2,
+ termination_policies=['OldestInstance', 'OldestLaunchConfiguration'])
+ self.service_connection.put_notification_configuration(autoscale, 'arn:aws:sns:us-east-1:19890506:AutoScaling-Up', ['autoscaling:EC2_INSTANCE_LAUNCH'])
+ self.assert_request_parameters({
+ 'Action': 'PutNotificationConfiguration',
+ 'AutoScalingGroupName': 'ana',
+ 'NotificationTypes.member.1': 'autoscaling:EC2_INSTANCE_LAUNCH',
+ 'TopicARN': 'arn:aws:sns:us-east-1:19890506:AutoScaling-Up',
+ }, ignore_params_values=['Version'])
+
+
+class TestDeleteNotificationConfiguration(AWSMockServiceTestCase):
+ connection_class = AutoScaleConnection
+
+ def setUp(self):
+ super(TestDeleteNotificationConfiguration, self).setUp()
+
+ def default_body(self):
+ return """
+ <DeleteNotificationConfigurationResponse>
+ <ResponseMetadata>
+ <RequestId>requestid</RequestId>
+ </ResponseMetadata>
+ </DeleteNotificationConfigurationResponse>
+ """
+
+ def test_autoscaling_group_put_notification_configuration(self):
+ self.set_http_response(status_code=200)
+ autoscale = AutoScalingGroup(
+ name='ana', launch_config='lauch_config',
+ min_size=1, max_size=2,
+ termination_policies=['OldestInstance', 'OldestLaunchConfiguration'])
+ self.service_connection.delete_notification_configuration(autoscale, 'arn:aws:sns:us-east-1:19890506:AutoScaling-Up')
+ self.assert_request_parameters({
+ 'Action': 'DeleteNotificationConfiguration',
+ 'AutoScalingGroupName': 'ana',
+ 'TopicARN': 'arn:aws:sns:us-east-1:19890506:AutoScaling-Up',
+ }, ignore_params_values=['Version'])
+
+class TestAutoScalingTag(AWSMockServiceTestCase):
+ connection_class = AutoScaleConnection
+
+ def default_body(self):
+ return """
+ <CreateOrUpdateTagsResponse>
+ <ResponseMetadata>
+ <RequestId>requestId</RequestId>
+ </ResponseMetadata>
+ </CreateOrUpdateTagsResponse>
+ """
+
+ def test_create_or_update_tags(self):
+ self.set_http_response(status_code=200)
+
+ tags = [
+ Tag(
+ connection=self.service_connection,
+ key='alpha',
+ value='tango',
+ resource_id='sg-00000000',
+ resource_type='auto-scaling-group',
+ propagate_at_launch=True
+ ),
+ Tag(
+ connection=self.service_connection,
+ key='bravo',
+ value='sierra',
+ resource_id='sg-00000000',
+ resource_type='auto-scaling-group',
+ propagate_at_launch=False
+ )]
+
+
+ response = self.service_connection.create_or_update_tags(tags)
+
+ self.assert_request_parameters({
+ 'Action': 'CreateOrUpdateTags',
+ 'Tags.member.1.ResourceType': 'auto-scaling-group',
+ 'Tags.member.1.ResourceId': 'sg-00000000',
+ 'Tags.member.1.Key': 'alpha',
+ 'Tags.member.1.Value': 'tango',
+ 'Tags.member.1.PropagateAtLaunch': 'true',
+ 'Tags.member.2.ResourceType': 'auto-scaling-group',
+ 'Tags.member.2.ResourceId': 'sg-00000000',
+ 'Tags.member.2.Key': 'bravo',
+ 'Tags.member.2.Value': 'sierra',
+ 'Tags.member.2.PropagateAtLaunch': 'false'
+ }, ignore_params_values=['Version'])
+
+ def test_endElement(self):
+ for i in [
+ ('Key', 'mykey', 'key'),
+ ('Value', 'myvalue', 'value'),
+ ('ResourceType', 'auto-scaling-group', 'resource_type'),
+ ('ResourceId', 'sg-01234567', 'resource_id'),
+ ('PropagateAtLaunch', 'true', 'propagate_at_launch')]:
+ self.check_tag_attributes_set(i[0], i[1], i[2])
+
+
+ def check_tag_attributes_set(self, name, value, attr):
+ tag = Tag()
+ tag.endElement(name, value, None)
+ if value == 'true':
+ self.assertEqual(getattr(tag, attr), True)
+ else:
+ self.assertEqual(getattr(tag, attr), value)
if __name__ == '__main__':
unittest.main()
diff --git a/tests/unit/ec2/elb/test_loadbalancer.py b/tests/unit/ec2/elb/test_loadbalancer.py
index d5e126c2..3577d7f8 100644
--- a/tests/unit/ec2/elb/test_loadbalancer.py
+++ b/tests/unit/ec2/elb/test_loadbalancer.py
@@ -6,6 +6,7 @@ from tests.unit import AWSMockServiceTestCase
import mock
from boto.ec2.elb import ELBConnection
+from boto.ec2.elb import LoadBalancer
DISABLE_RESPONSE = r"""<?xml version="1.0" encoding="UTF-8"?>
<DisableAvailabilityZonesForLoadBalancerResult xmlns="http://ec2.amazonaws.com/doc/2013-02-01/">
@@ -29,5 +30,101 @@ class TestInstanceStatusResponseParsing(unittest.TestCase):
self.assertEqual(disabled, ['sample-zone'])
+DESCRIBE_RESPONSE = r"""<?xml version="1.0" encoding="UTF-8"?>
+<DescribeLoadBalancersResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2012-06-01/">
+ <DescribeLoadBalancersResult>
+ <LoadBalancerDescriptions>
+ <member>
+ <SecurityGroups/>
+ <CreatedTime>2013-07-09T19:18:00.520Z</CreatedTime>
+ <LoadBalancerName>elb-boto-unit-test</LoadBalancerName>
+ <HealthCheck/>
+ <ListenerDescriptions>
+ <member>
+ <PolicyNames/>
+ <Listener/>
+ </member>
+ </ListenerDescriptions>
+ <Instances/>
+ <Policies>
+ <AppCookieStickinessPolicies/>
+ <OtherPolicies>
+ <member>AWSConsole-SSLNegotiationPolicy-my-test-loadbalancer</member>
+ <member>EnableProxyProtocol</member>
+ </OtherPolicies>
+ <LBCookieStickinessPolicies/>
+ </Policies>
+ <AvailabilityZones>
+ <member>us-east-1a</member>
+ </AvailabilityZones>
+ <CanonicalHostedZoneName>elb-boto-unit-test-408121642.us-east-1.elb.amazonaws.com</CanonicalHostedZoneName>
+ <CanonicalHostedZoneNameID>Z3DZXE0Q79N41H</CanonicalHostedZoneNameID>
+ <Scheme>internet-facing</Scheme>
+ <SourceSecurityGroup>
+ <OwnerAlias>amazon-elb</OwnerAlias>
+ <GroupName>amazon-elb-sg</GroupName>
+ </SourceSecurityGroup>
+ <DNSName>elb-boto-unit-test-408121642.us-east-1.elb.amazonaws.com</DNSName>
+ <BackendServerDescriptions>
+ <member>
+ <PolicyNames>
+ <member>EnableProxyProtocol</member>
+ </PolicyNames>
+ <InstancePort>80</InstancePort>
+ </member>
+ </BackendServerDescriptions>
+ <Subnets/>
+ </member>
+ </LoadBalancerDescriptions>
+ </DescribeLoadBalancersResult>
+ <ResponseMetadata>
+ <RequestId>5763d932-e8cc-11e2-a940-11136cceffb8</RequestId>
+ </ResponseMetadata>
+</DescribeLoadBalancersResponse>
+"""
+
+class TestDescribeLoadBalancers(unittest.TestCase):
+ def test_other_policy(self):
+ elb = ELBConnection(aws_access_key_id='aws_access_key_id',
+ aws_secret_access_key='aws_secret_access_key')
+ mock_response = mock.Mock()
+ mock_response.read.return_value = DESCRIBE_RESPONSE
+ mock_response.status = 200
+ elb.make_request = mock.Mock(return_value=mock_response)
+ load_balancers = elb.get_all_load_balancers()
+ self.assertEqual(len(load_balancers), 1)
+
+ lb = load_balancers[0]
+ self.assertEqual(len(lb.policies.other_policies), 2)
+ self.assertEqual(lb.policies.other_policies[0].policy_name,
+ 'AWSConsole-SSLNegotiationPolicy-my-test-loadbalancer')
+ self.assertEqual(lb.policies.other_policies[1].policy_name,
+ 'EnableProxyProtocol')
+
+ self.assertEqual(len(lb.backends), 1)
+ self.assertEqual(len(lb.backends[0].policies), 1)
+ self.assertEqual(lb.backends[0].policies[0].policy_name,
+ 'EnableProxyProtocol')
+ self.assertEqual(lb.backends[0].instance_port, 80)
+
+
+DETACH_RESPONSE = r"""<?xml version="1.0" encoding="UTF-8"?>
+<DetachLoadBalancerFromSubnets xmlns="http://ec2.amazonaws.com/doc/2013-02-01/">
+ <requestId>3be1508e-c444-4fef-89cc-0b1223c4f02fEXAMPLE</requestId>
+</DetachLoadBalancerFromSubnets>
+"""
+
+class TestDetachSubnets(unittest.TestCase):
+ def test_detach_subnets(self):
+ elb = ELBConnection(aws_access_key_id='aws_access_key_id',
+ aws_secret_access_key='aws_secret_access_key')
+ lb = LoadBalancer(elb, "mylb")
+
+ mock_response = mock.Mock()
+ mock_response.read.return_value = DETACH_RESPONSE
+ mock_response.status = 200
+ elb.make_request = mock.Mock(return_value=mock_response)
+ lb.detach_subnets("s-xxx")
+
if __name__ == '__main__':
unittest.main()
diff --git a/tests/unit/ec2/test_address.py b/tests/unit/ec2/test_address.py
index f2661979..765ce422 100644
--- a/tests/unit/ec2/test_address.py
+++ b/tests/unit/ec2/test_address.py
@@ -25,15 +25,25 @@ class AddressTest(unittest.TestCase):
def test_release_calls_connection_release_address_with_correct_args(self):
self.address.release()
- self.address.connection.release_address.assert_called_with("192.168.1.1")
+ self.address.connection.release_address.assert_called_with(
+ "192.168.1.1",
+ dry_run=False
+ )
def test_associate_calls_connection_associate_address_with_correct_args(self):
self.address.associate(1)
- self.address.connection.associate_address.assert_called_with(1, "192.168.1.1")
+ self.address.connection.associate_address.assert_called_with(
+ 1,
+ "192.168.1.1",
+ dry_run=False
+ )
def test_disassociate_calls_connection_disassociate_address_with_correct_args(self):
self.address.disassociate()
- self.address.connection.disassociate_address.assert_called_with("192.168.1.1")
+ self.address.connection.disassociate_address.assert_called_with(
+ "192.168.1.1",
+ dry_run=False
+ )
if __name__ == "__main__":
unittest.main()
diff --git a/tests/unit/ec2/test_blockdevicemapping.py b/tests/unit/ec2/test_blockdevicemapping.py
index 02ecf582..78539744 100644
--- a/tests/unit/ec2/test_blockdevicemapping.py
+++ b/tests/unit/ec2/test_blockdevicemapping.py
@@ -1,8 +1,12 @@
import mock
import unittest
+from boto.ec2.connection import EC2Connection
from boto.ec2.blockdevicemapping import BlockDeviceType, BlockDeviceMapping
+from tests.unit import AWSMockServiceTestCase
+
+
class BlockDeviceTypeTests(unittest.TestCase):
def setUp(self):
self.block_device_type = BlockDeviceType()
@@ -75,5 +79,55 @@ class BlockDeviceMappingTests(unittest.TestCase):
self.block_device_mapping.endElement("item", "some item", None)
self.assertEqual(self.block_device_mapping["some name"], "some value")
+
+class TestLaunchConfiguration(AWSMockServiceTestCase):
+ connection_class = EC2Connection
+
+ def default_body(self):
+ # This is a dummy response
+ return """
+ <DescribeLaunchConfigurationsResponse>
+ </DescribeLaunchConfigurationsResponse>
+ """
+
+ def test_run_instances_block_device_mapping(self):
+ # Same as the test in ``unit/ec2/autoscale/test_group.py:TestLaunchConfiguration``,
+ # but with modified request parameters (due to a mismatch between EC2 &
+ # Autoscaling).
+ self.set_http_response(status_code=200)
+ dev_sdf = BlockDeviceType(snapshot_id='snap-12345')
+ dev_sdg = BlockDeviceType(snapshot_id='snap-12346')
+
+ bdm = BlockDeviceMapping()
+ bdm['/dev/sdf'] = dev_sdf
+ bdm['/dev/sdg'] = dev_sdg
+
+ response = self.service_connection.run_instances(
+ image_id='123456',
+ instance_type='m1.large',
+ security_groups=['group1', 'group2'],
+ block_device_map=bdm
+ )
+
+ self.assert_request_parameters({
+ 'Action': 'RunInstances',
+ 'BlockDeviceMapping.1.DeviceName': '/dev/sdf',
+ 'BlockDeviceMapping.1.Ebs.DeleteOnTermination': 'false',
+ 'BlockDeviceMapping.1.Ebs.SnapshotId': 'snap-12345',
+ 'BlockDeviceMapping.2.DeviceName': '/dev/sdg',
+ 'BlockDeviceMapping.2.Ebs.DeleteOnTermination': 'false',
+ 'BlockDeviceMapping.2.Ebs.SnapshotId': 'snap-12346',
+ 'ImageId': '123456',
+ 'InstanceType': 'm1.large',
+ 'MaxCount': 1,
+ 'MinCount': 1,
+ 'SecurityGroup.1': 'group1',
+ 'SecurityGroup.2': 'group2',
+ }, ignore_params_values=[
+ 'Version', 'AWSAccessKeyId', 'SignatureMethod', 'SignatureVersion',
+ 'Timestamp'
+ ])
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/unit/ec2/test_connection.py b/tests/unit/ec2/test_connection.py
index d06288dc..deb6759c 100644
--- a/tests/unit/ec2/test_connection.py
+++ b/tests/unit/ec2/test_connection.py
@@ -1,8 +1,17 @@
#!/usr/bin/env python
+import httplib
+
+from datetime import datetime, timedelta
+from mock import MagicMock, Mock, patch
from tests.unit import unittest
from tests.unit import AWSMockServiceTestCase
+import boto.ec2
+
+from boto.regioninfo import RegionInfo
from boto.ec2.connection import EC2Connection
+from boto.ec2.snapshot import Snapshot
+from boto.ec2.reservedinstance import ReservedInstancesConfiguration
class TestEC2ConnectionBase(AWSMockServiceTestCase):
@@ -475,6 +484,47 @@ class TestCopySnapshot(TestEC2ConnectionBase):
'SignatureVersion', 'Timestamp',
'Version'])
+class TestCopyImage(TestEC2ConnectionBase):
+ def default_body(self):
+ return """
+ <CopyImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-07-15/">
+ <requestId>request_id</requestId>
+ <imageId>ami-copied-id</imageId>
+ </CopyImageResponse>
+ """
+
+ def test_copy_image(self):
+ self.set_http_response(status_code=200)
+ copied_ami = self.ec2.copy_image('us-west-2', 'ami-id',
+ 'name', 'description', 'client-token')
+ self.assertEqual(copied_ami.image_id, 'ami-copied-id')
+
+ self.assert_request_parameters({
+ 'Action': 'CopyImage',
+ 'Description': 'description',
+ 'Name': 'name',
+ 'SourceRegion': 'us-west-2',
+ 'SourceImageId': 'ami-id',
+ 'ClientToken': 'client-token'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+ def test_copy_image_without_name(self):
+ self.set_http_response(status_code=200)
+ copied_ami = self.ec2.copy_image('us-west-2', 'ami-id',
+ description='description',
+ client_token='client-token')
+ self.assertEqual(copied_ami.image_id, 'ami-copied-id')
+
+ self.assert_request_parameters({
+ 'Action': 'CopyImage',
+ 'Description': 'description',
+ 'SourceRegion': 'us-west-2',
+ 'SourceImageId': 'ami-id',
+ 'ClientToken': 'client-token'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
class TestAccountAttributes(TestEC2ConnectionBase):
def default_body(self):
@@ -562,5 +612,612 @@ class TestDescribeVPCAttribute(TestEC2ConnectionBase):
'Version'])
+class TestGetAllNetworkInterfaces(TestEC2ConnectionBase):
+ def default_body(self):
+ return """
+<DescribeNetworkInterfacesResponse xmlns="http://ec2.amazonaws.com/\
+ doc/2013-06-15/">
+ <requestId>fc45294c-006b-457b-bab9-012f5b3b0e40</requestId>
+ <networkInterfaceSet>
+ <item>
+ <networkInterfaceId>eni-0f62d866</networkInterfaceId>
+ <subnetId>subnet-c53c87ac</subnetId>
+ <vpcId>vpc-cc3c87a5</vpcId>
+ <availabilityZone>ap-southeast-1b</availabilityZone>
+ <description/>
+ <ownerId>053230519467</ownerId>
+ <requesterManaged>false</requesterManaged>
+ <status>in-use</status>
+ <macAddress>02:81:60:cb:27:37</macAddress>
+ <privateIpAddress>10.0.0.146</privateIpAddress>
+ <sourceDestCheck>true</sourceDestCheck>
+ <groupSet>
+ <item>
+ <groupId>sg-3f4b5653</groupId>
+ <groupName>default</groupName>
+ </item>
+ </groupSet>
+ <attachment>
+ <attachmentId>eni-attach-6537fc0c</attachmentId>
+ <instanceId>i-22197876</instanceId>
+ <instanceOwnerId>053230519467</instanceOwnerId>
+ <deviceIndex>5</deviceIndex>
+ <status>attached</status>
+ <attachTime>2012-07-01T21:45:27.000Z</attachTime>
+ <deleteOnTermination>true</deleteOnTermination>
+ </attachment>
+ <tagSet/>
+ <privateIpAddressesSet>
+ <item>
+ <privateIpAddress>10.0.0.146</privateIpAddress>
+ <primary>true</primary>
+ </item>
+ <item>
+ <privateIpAddress>10.0.0.148</privateIpAddress>
+ <primary>false</primary>
+ </item>
+ <item>
+ <privateIpAddress>10.0.0.150</privateIpAddress>
+ <primary>false</primary>
+ </item>
+ </privateIpAddressesSet>
+ </item>
+ </networkInterfaceSet>
+</DescribeNetworkInterfacesResponse>"""
+
+ def test_attachment_has_device_index(self):
+ self.set_http_response(status_code=200)
+ parsed = self.ec2.get_all_network_interfaces()
+
+ self.assertEqual(5, parsed[0].attachment.device_index)
+
+class TestGetAllImages(TestEC2ConnectionBase):
+ def default_body(self):
+ return """
+<DescribeImagesResponse xmlns="http://ec2.amazonaws.com/doc/2013-02-01/">
+ <requestId>e32375e8-4ac3-4099-a8bf-3ec902b9023e</requestId>
+ <imagesSet>
+ <item>
+ <imageId>ami-abcd1234</imageId>
+ <imageLocation>111111111111/windows2008r2-hvm-i386-20130702</imageLocation>
+ <imageState>available</imageState>
+ <imageOwnerId>111111111111</imageOwnerId>
+ <isPublic>false</isPublic>
+ <architecture>i386</architecture>
+ <imageType>machine</imageType>
+ <platform>windows</platform>
+ <viridianEnabled>true</viridianEnabled>
+ <name>Windows Test</name>
+ <description>Windows Test Description</description>
+ <billingProducts>
+ <item>
+ <billingProduct>bp-6ba54002</billingProduct>
+ </item>
+ </billingProducts>
+ <rootDeviceType>ebs</rootDeviceType>
+ <rootDeviceName>/dev/sda1</rootDeviceName>
+ <blockDeviceMapping>
+ <item>
+ <deviceName>/dev/sda1</deviceName>
+ <ebs>
+ <snapshotId>snap-abcd1234</snapshotId>
+ <volumeSize>30</volumeSize>
+ <deleteOnTermination>true</deleteOnTermination>
+ <volumeType>standard</volumeType>
+ </ebs>
+ </item>
+ <item>
+ <deviceName>xvdb</deviceName>
+ <virtualName>ephemeral0</virtualName>
+ </item>
+ <item>
+ <deviceName>xvdc</deviceName>
+ <virtualName>ephemeral1</virtualName>
+ </item>
+ <item>
+ <deviceName>xvdd</deviceName>
+ <virtualName>ephemeral2</virtualName>
+ </item>
+ <item>
+ <deviceName>xvde</deviceName>
+ <virtualName>ephemeral3</virtualName>
+ </item>
+ </blockDeviceMapping>
+ <virtualizationType>hvm</virtualizationType>
+ <hypervisor>xen</hypervisor>
+ </item>
+ </imagesSet>
+</DescribeImagesResponse>"""
+
+ def test_get_all_images(self):
+ self.set_http_response(status_code=200)
+ parsed = self.ec2.get_all_images()
+ self.assertEquals(1, len(parsed))
+ self.assertEquals("ami-abcd1234", parsed[0].id)
+ self.assertEquals("111111111111/windows2008r2-hvm-i386-20130702", parsed[0].location)
+ self.assertEquals("available", parsed[0].state)
+ self.assertEquals("111111111111", parsed[0].ownerId)
+ self.assertEquals("111111111111", parsed[0].owner_id)
+ self.assertEquals(False, parsed[0].is_public)
+ self.assertEquals("i386", parsed[0].architecture)
+ self.assertEquals("machine", parsed[0].type)
+ self.assertEquals(None, parsed[0].kernel_id)
+ self.assertEquals(None, parsed[0].ramdisk_id)
+ self.assertEquals(None, parsed[0].owner_alias)
+ self.assertEquals("windows", parsed[0].platform)
+ self.assertEquals("Windows Test", parsed[0].name)
+ self.assertEquals("Windows Test Description", parsed[0].description)
+ self.assertEquals("ebs", parsed[0].root_device_type)
+ self.assertEquals("/dev/sda1", parsed[0].root_device_name)
+ self.assertEquals("hvm", parsed[0].virtualization_type)
+ self.assertEquals("xen", parsed[0].hypervisor)
+ self.assertEquals(None, parsed[0].instance_lifecycle)
+
+ # 1 billing product parsed into a list
+ self.assertEquals(1, len(parsed[0].billing_products))
+ self.assertEquals("bp-6ba54002", parsed[0].billing_products[0])
+
+ # Just verify length, there is already a block_device_mapping test
+ self.assertEquals(5, len(parsed[0].block_device_mapping))
+
+ # TODO: No tests for product codes?
+
+
+class TestModifyInterfaceAttribute(TestEC2ConnectionBase):
+ def default_body(self):
+ return """
+<ModifyNetworkInterfaceAttributeResponse \
+ xmlns="http://ec2.amazonaws.com/doc/2013-06-15/">
+ <requestId>657a4623-5620-4232-b03b-427e852d71cf</requestId>
+ <return>true</return>
+</ModifyNetworkInterfaceAttributeResponse>
+"""
+
+ def test_modify_description(self):
+ self.set_http_response(status_code=200)
+ self.ec2.modify_network_interface_attribute('id', 'description', 'foo')
+
+ self.assert_request_parameters({
+ 'Action': 'ModifyNetworkInterfaceAttribute',
+ 'NetworkInterfaceId': 'id',
+ 'Description.Value': 'foo'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+
+ def test_modify_source_dest_check_bool(self):
+ self.set_http_response(status_code=200)
+ self.ec2.modify_network_interface_attribute('id', 'sourceDestCheck',
+ True)
+
+ self.assert_request_parameters({
+ 'Action': 'ModifyNetworkInterfaceAttribute',
+ 'NetworkInterfaceId': 'id',
+ 'SourceDestCheck.Value': 'true'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+
+ def test_modify_source_dest_check_str(self):
+ self.set_http_response(status_code=200)
+ self.ec2.modify_network_interface_attribute('id', 'sourceDestCheck',
+ 'true')
+
+ self.assert_request_parameters({
+ 'Action': 'ModifyNetworkInterfaceAttribute',
+ 'NetworkInterfaceId': 'id',
+ 'SourceDestCheck.Value': 'true'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+
+ def test_modify_source_dest_check_invalid(self):
+ self.set_http_response(status_code=200)
+
+ with self.assertRaises(ValueError):
+ self.ec2.modify_network_interface_attribute('id',
+ 'sourceDestCheck',
+ 123)
+
+ def test_modify_delete_on_termination_str(self):
+ self.set_http_response(status_code=200)
+ self.ec2.modify_network_interface_attribute('id',
+ 'deleteOnTermination',
+ True, attachment_id='bar')
+
+ self.assert_request_parameters({
+ 'Action': 'ModifyNetworkInterfaceAttribute',
+ 'NetworkInterfaceId': 'id',
+ 'Attachment.AttachmentId': 'bar',
+ 'Attachment.DeleteOnTermination': 'true'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+
+ def test_modify_delete_on_termination_bool(self):
+ self.set_http_response(status_code=200)
+ self.ec2.modify_network_interface_attribute('id',
+ 'deleteOnTermination',
+ 'false',
+ attachment_id='bar')
+
+ self.assert_request_parameters({
+ 'Action': 'ModifyNetworkInterfaceAttribute',
+ 'NetworkInterfaceId': 'id',
+ 'Attachment.AttachmentId': 'bar',
+ 'Attachment.DeleteOnTermination': 'false'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+
+ def test_modify_delete_on_termination_invalid(self):
+ self.set_http_response(status_code=200)
+
+ with self.assertRaises(ValueError):
+ self.ec2.modify_network_interface_attribute('id',
+ 'deleteOnTermination',
+ 123,
+ attachment_id='bar')
+
+ def test_modify_group_set_list(self):
+ self.set_http_response(status_code=200)
+ self.ec2.modify_network_interface_attribute('id', 'groupSet',
+ ['sg-1', 'sg-2'])
+
+ self.assert_request_parameters({
+ 'Action': 'ModifyNetworkInterfaceAttribute',
+ 'NetworkInterfaceId': 'id',
+ 'SecurityGroupId.1': 'sg-1',
+ 'SecurityGroupId.2': 'sg-2'},
+ ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'])
+
+ def test_modify_group_set_invalid(self):
+ self.set_http_response(status_code=200)
+
+ with self.assertRaisesRegexp(TypeError, 'iterable'):
+ self.ec2.modify_network_interface_attribute('id', 'groupSet',
+ False)
+
+ def test_modify_attr_invalid(self):
+ self.set_http_response(status_code=200)
+
+ with self.assertRaisesRegexp(ValueError, 'Unknown attribute'):
+ self.ec2.modify_network_interface_attribute('id', 'invalid', 0)
+
+
+class TestConnectToRegion(unittest.TestCase):
+ def setUp(self):
+ self.https_connection = Mock(spec=httplib.HTTPSConnection)
+ self.https_connection_factory = (
+ Mock(return_value=self.https_connection), ())
+
+ def test_aws_region(self):
+ region = boto.ec2.RegionData.keys()[0]
+ self.ec2 = boto.ec2.connect_to_region(region,
+ https_connection_factory=self.https_connection_factory,
+ aws_access_key_id='aws_access_key_id',
+ aws_secret_access_key='aws_secret_access_key'
+ )
+ self.assertEqual(boto.ec2.RegionData[region], self.ec2.host)
+
+ def test_non_aws_region(self):
+ self.ec2 = boto.ec2.connect_to_region('foo',
+ https_connection_factory=self.https_connection_factory,
+ aws_access_key_id='aws_access_key_id',
+ aws_secret_access_key='aws_secret_access_key',
+ region = RegionInfo(name='foo', endpoint='https://foo.com/bar')
+ )
+ self.assertEqual('https://foo.com/bar', self.ec2.host)
+
+ def test_missing_region(self):
+ self.ec2 = boto.ec2.connect_to_region('foo',
+ https_connection_factory=self.https_connection_factory,
+ aws_access_key_id='aws_access_key_id',
+ aws_secret_access_key='aws_secret_access_key'
+ )
+ self.assertEqual(None, self.ec2)
+
+
+class TestTrimSnapshots(TestEC2ConnectionBase):
+ """
+ Test snapshot trimming functionality by ensuring that expected calls
+ are made when given a known set of volume snapshots.
+ """
+ def _get_snapshots(self):
+ """
+ Generate a list of fake snapshots with names and dates.
+ """
+ snaps = []
+
+ # Generate some dates offset by days, weeks, months
+ now = datetime.now()
+ dates = [
+ now,
+ now - timedelta(days=1),
+ now - timedelta(days=2),
+ now - timedelta(days=7),
+ now - timedelta(days=14),
+ datetime(now.year, now.month, 1) - timedelta(days=30),
+ datetime(now.year, now.month, 1) - timedelta(days=60),
+ datetime(now.year, now.month, 1) - timedelta(days=90)
+ ]
+
+ for date in dates:
+ # Create a fake snapshot for each date
+ snap = Snapshot(self.ec2)
+ snap.tags['Name'] = 'foo'
+ # Times are expected to be ISO8601 strings
+ snap.start_time = date.strftime('%Y-%m-%dT%H:%M:%S.000Z')
+ snaps.append(snap)
+
+ return snaps
+
+ def test_trim_defaults(self):
+ """
+ Test trimming snapshots with the default arguments, which should
+ keep all monthly backups forever. The result of this test should
+ be that nothing is deleted.
+ """
+ # Setup mocks
+ orig = {
+ 'get_all_snapshots': self.ec2.get_all_snapshots,
+ 'delete_snapshot': self.ec2.delete_snapshot
+ }
+
+ snaps = self._get_snapshots()
+
+ self.ec2.get_all_snapshots = MagicMock(return_value=snaps)
+ self.ec2.delete_snapshot = MagicMock()
+
+ # Call the tested method
+ self.ec2.trim_snapshots()
+
+ # Assertions
+ self.assertEqual(True, self.ec2.get_all_snapshots.called)
+ self.assertEqual(False, self.ec2.delete_snapshot.called)
+
+ # Restore
+ self.ec2.get_all_snapshots = orig['get_all_snapshots']
+ self.ec2.delete_snapshot = orig['delete_snapshot']
+
+ def test_trim_months(self):
+ """
+ Test trimming monthly snapshots and ensure that older months
+ get deleted properly. The result of this test should be that
+ the two oldest snapshots get deleted.
+ """
+ # Setup mocks
+ orig = {
+ 'get_all_snapshots': self.ec2.get_all_snapshots,
+ 'delete_snapshot': self.ec2.delete_snapshot
+ }
+
+ snaps = self._get_snapshots()
+
+ self.ec2.get_all_snapshots = MagicMock(return_value=snaps)
+ self.ec2.delete_snapshot = MagicMock()
+
+ # Call the tested method
+ self.ec2.trim_snapshots(monthly_backups=1)
+
+ # Assertions
+ self.assertEqual(True, self.ec2.get_all_snapshots.called)
+ self.assertEqual(2, self.ec2.delete_snapshot.call_count)
+
+ # Restore
+ self.ec2.get_all_snapshots = orig['get_all_snapshots']
+ self.ec2.delete_snapshot = orig['delete_snapshot']
+
+
+class TestModifyReservedInstances(TestEC2ConnectionBase):
+ def default_body(self):
+ return """<ModifyReservedInstancesResponse xmlns='http://ec2.amazonaws.com/doc/2013-08-15/'>
+ <requestId>bef729b6-0731-4489-8881-2258746ae163</requestId>
+ <reservedInstancesModificationId>rimod-3aae219d-3d63-47a9-a7e9-e764example</reservedInstancesModificationId>
+</ModifyReservedInstancesResponse>"""
+
+ def test_serialized_api_args(self):
+ self.set_http_response(status_code=200)
+ response = self.ec2.modify_reserved_instances(
+ 'a-token-goes-here',
+ reserved_instance_ids=[
+ '2567o137-8a55-48d6-82fb-7258506bb497',
+ ],
+ target_configurations=[
+ ReservedInstancesConfiguration(
+ availability_zone='us-west-2c',
+ platform='EC2-VPC',
+ instance_count=3
+ ),
+ ]
+ )
+ self.assert_request_parameters({
+ 'Action': 'ModifyReservedInstances',
+ 'ClientToken': 'a-token-goes-here',
+ 'ReservedInstancesConfigurationSetItemType.0.AvailabilityZone': 'us-west-2c',
+ 'ReservedInstancesConfigurationSetItemType.0.InstanceCount': 3,
+ 'ReservedInstancesConfigurationSetItemType.0.Platform': 'EC2-VPC',
+ 'ReservedInstancesId.1': '2567o137-8a55-48d6-82fb-7258506bb497'
+ }, ignore_params_values=[
+ 'AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'
+ ])
+
+ self.assertEqual(response, 'rimod-3aae219d-3d63-47a9-a7e9-e764example')
+
+
+class TestDescribeReservedInstancesModifications(TestEC2ConnectionBase):
+ def default_body(self):
+ return """<DescribeReservedInstancesModificationsResponse xmlns='http://ec2.amazonaws.com/doc/2013-08-15/'>
+ <requestId>eb4a6e3c-3689-445c-b536-19e38df35898</requestId>
+ <reservedInstancesModificationsSet>
+ <item>
+ <reservedInstancesModificationId>rimod-49b9433e-fdc7-464a-a6e5-9dabcexample</reservedInstancesModificationId>
+ <reservedInstancesSet>
+ <item>
+ <reservedInstancesId>2567o137-8a55-48d6-82fb-7258506bb497</reservedInstancesId>
+ </item>
+ </reservedInstancesSet>
+ <modificationResultSet>
+ <item>
+ <reservedInstancesId>9d5cb137-5d65-4479-b4ac-8c337example</reservedInstancesId>
+ <targetConfiguration>
+ <availabilityZone>us-east-1b</availabilityZone>
+ <platform>EC2-VPC</platform>
+ <instanceCount>1</instanceCount>
+ </targetConfiguration>
+ </item>
+ </modificationResultSet>
+ <createDate>2013-09-02T21:20:19.637Z</createDate>
+ <updateDate>2013-09-02T21:38:24.143Z</updateDate>
+ <effectiveDate>2013-09-02T21:00:00.000Z</effectiveDate>
+ <status>fulfilled</status>
+ <clientToken>token-f5b56c05-09b0-4d17-8d8c-c75d8a67b806</clientToken>
+ </item>
+ </reservedInstancesModificationsSet>
+</DescribeReservedInstancesModificationsResponse>"""
+
+ def test_serialized_api_args(self):
+ self.set_http_response(status_code=200)
+ response = self.ec2.describe_reserved_instances_modifications(
+ reserved_instances_modification_ids=[
+ '2567o137-8a55-48d6-82fb-7258506bb497'
+ ],
+ filters={
+ 'status': 'processing',
+ }
+ )
+ self.assert_request_parameters({
+ 'Action': 'DescribeReservedInstancesModifications',
+ 'Filter.1.Name': 'status',
+ 'Filter.1.Value.1': 'processing',
+ 'ReservedInstancesModificationId.1': '2567o137-8a55-48d6-82fb-7258506bb497'
+ }, ignore_params_values=[
+ 'AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'
+ ])
+
+ # Make sure the response was parsed correctly.
+ self.assertEqual(
+ response[0].modification_id,
+ 'rimod-49b9433e-fdc7-464a-a6e5-9dabcexample'
+ )
+ self.assertEqual(
+ response[0].create_date,
+ datetime(2013, 9, 2, 21, 20, 19, 637000)
+ )
+ self.assertEqual(
+ response[0].update_date,
+ datetime(2013, 9, 2, 21, 38, 24, 143000)
+ )
+ self.assertEqual(
+ response[0].effective_date,
+ datetime(2013, 9, 2, 21, 0, 0, 0)
+ )
+ self.assertEqual(
+ response[0].status,
+ 'fulfilled'
+ )
+ self.assertEqual(
+ response[0].status_message,
+ None
+ )
+ self.assertEqual(
+ response[0].client_token,
+ 'token-f5b56c05-09b0-4d17-8d8c-c75d8a67b806'
+ )
+ self.assertEqual(
+ response[0].reserved_instances[0].id,
+ '2567o137-8a55-48d6-82fb-7258506bb497'
+ )
+ self.assertEqual(
+ response[0].modification_results[0].availability_zone,
+ 'us-east-1b'
+ )
+ self.assertEqual(
+ response[0].modification_results[0].platform,
+ 'EC2-VPC'
+ )
+ self.assertEqual(
+ response[0].modification_results[0].instance_count,
+ 1
+ )
+ self.assertEqual(len(response), 1)
+
+
+class TestRegisterImage(TestEC2ConnectionBase):
+ def default_body(self):
+ return """
+ <RegisterImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-08-15/">
+ <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
+ <imageId>ami-1a2b3c4d</imageId>
+ </RegisterImageResponse>
+ """
+
+ def test_vm_type_default(self):
+ self.set_http_response(status_code=200)
+ self.ec2.register_image('name', 'description',
+ image_location='s3://foo')
+
+ self.assert_request_parameters({
+ 'Action': 'RegisterImage',
+ 'ImageLocation': 's3://foo',
+ 'Name': 'name',
+ 'Description': 'description',
+ }, ignore_params_values=[
+ 'AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'
+ ])
+
+ def test_vm_type_hvm(self):
+ self.set_http_response(status_code=200)
+ self.ec2.register_image('name', 'description',
+ image_location='s3://foo',
+ virtualization_type='hvm')
+
+ self.assert_request_parameters({
+ 'Action': 'RegisterImage',
+ 'ImageLocation': 's3://foo',
+ 'Name': 'name',
+ 'Description': 'description',
+ 'VirtualizationType': 'hvm'
+ }, ignore_params_values=[
+ 'AWSAccessKeyId', 'SignatureMethod',
+ 'SignatureVersion', 'Timestamp',
+ 'Version'
+ ])
+
+
+class TestTerminateInstances(TestEC2ConnectionBase):
+ def default_body(self):
+ return """<?xml version="1.0" ?>
+ <TerminateInstancesResponse xmlns="http://ec2.amazonaws.com/doc/2013-07-15/">
+ <requestId>req-59a9ad52-0434-470c-ad48-4f89ded3a03e</requestId>
+ <instancesSet>
+ <item>
+ <instanceId>i-000043a2</instanceId>
+ <shutdownState>
+ <code>16</code>
+ <name>running</name>
+ </shutdownState>
+ <previousState>
+ <code>16</code>
+ <name>running</name>
+ </previousState>
+ </item>
+ </instancesSet>
+ </TerminateInstancesResponse>
+ """
+
+ def test_terminate_bad_response(self):
+ self.set_http_response(status_code=200)
+ self.ec2.terminate_instances('foo')
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/tests/unit/ec2/test_instance.py b/tests/unit/ec2/test_instance.py
index c48ef114..6ee0f2f2 100644
--- a/tests/unit/ec2/test_instance.py
+++ b/tests/unit/ec2/test_instance.py
@@ -216,7 +216,7 @@ class TestDescribeInstances(AWSMockServiceTestCase):
def test_multiple_private_ip_addresses(self):
self.set_http_response(status_code=200)
- api_response = self.service_connection.get_all_instances()
+ api_response = self.service_connection.get_all_reservations()
self.assertEqual(len(api_response), 1)
instances = api_response[0].instances
diff --git a/tests/unit/ec2/test_networkinterface.py b/tests/unit/ec2/test_networkinterface.py
index b23f6c36..81fa4aef 100644
--- a/tests/unit/ec2/test_networkinterface.py
+++ b/tests/unit/ec2/test_networkinterface.py
@@ -23,7 +23,7 @@
from tests.unit import unittest
-
+from boto.exception import BotoClientError
from boto.ec2.networkinterface import NetworkInterfaceCollection
from boto.ec2.networkinterface import NetworkInterfaceSpecification
from boto.ec2.networkinterface import PrivateIPAddress
@@ -42,7 +42,8 @@ class TestNetworkInterfaceCollection(unittest.TestCase):
description='description1',
private_ip_address='10.0.0.54', delete_on_termination=False,
private_ip_addresses=[self.private_ip_address1,
- self.private_ip_address2])
+ self.private_ip_address2]
+ )
self.private_ip_address3 = PrivateIPAddress(
private_ip_address='10.0.1.10', primary=False)
@@ -54,7 +55,18 @@ class TestNetworkInterfaceCollection(unittest.TestCase):
groups=['group_id1', 'group_id2'],
private_ip_address='10.0.1.54', delete_on_termination=False,
private_ip_addresses=[self.private_ip_address3,
- self.private_ip_address4])
+ self.private_ip_address4]
+ )
+
+ self.network_interfaces_spec3 = NetworkInterfaceSpecification(
+ device_index=0, subnet_id='subnet_id2',
+ description='description2',
+ groups=['group_id1', 'group_id2'],
+ private_ip_address='10.0.1.54', delete_on_termination=False,
+ private_ip_addresses=[self.private_ip_address3,
+ self.private_ip_address4],
+ associate_public_ip_address=True
+ )
def test_param_serialization(self):
collection = NetworkInterfaceCollection(self.network_interfaces_spec1,
@@ -62,34 +74,33 @@ class TestNetworkInterfaceCollection(unittest.TestCase):
params = {}
collection.build_list_params(params)
self.assertDictEqual(params, {
- 'NetworkInterface.1.DeviceIndex': '1',
- 'NetworkInterface.1.DeleteOnTermination': 'false',
- 'NetworkInterface.1.Description': 'description1',
- 'NetworkInterface.1.PrivateIpAddress': '10.0.0.54',
- 'NetworkInterface.1.SubnetId': 'subnet_id',
- 'NetworkInterface.1.PrivateIpAddresses.1.Primary': 'false',
- 'NetworkInterface.1.PrivateIpAddresses.1.PrivateIpAddress':
+ 'NetworkInterface.0.DeviceIndex': '1',
+ 'NetworkInterface.0.DeleteOnTermination': 'false',
+ 'NetworkInterface.0.Description': 'description1',
+ 'NetworkInterface.0.PrivateIpAddress': '10.0.0.54',
+ 'NetworkInterface.0.SubnetId': 'subnet_id',
+ 'NetworkInterface.0.PrivateIpAddresses.0.Primary': 'false',
+ 'NetworkInterface.0.PrivateIpAddresses.0.PrivateIpAddress':
'10.0.0.10',
- 'NetworkInterface.1.PrivateIpAddresses.2.Primary': 'false',
- 'NetworkInterface.1.PrivateIpAddresses.2.PrivateIpAddress':
+ 'NetworkInterface.0.PrivateIpAddresses.1.Primary': 'false',
+ 'NetworkInterface.0.PrivateIpAddresses.1.PrivateIpAddress':
'10.0.0.11',
- 'NetworkInterface.2.DeviceIndex': '2',
- 'NetworkInterface.2.Description': 'description2',
- 'NetworkInterface.2.DeleteOnTermination': 'false',
- 'NetworkInterface.2.PrivateIpAddress': '10.0.1.54',
- 'NetworkInterface.2.SubnetId': 'subnet_id2',
- 'NetworkInterface.2.SecurityGroupId.1': 'group_id1',
- 'NetworkInterface.2.SecurityGroupId.2': 'group_id2',
- 'NetworkInterface.2.PrivateIpAddresses.1.Primary': 'false',
- 'NetworkInterface.2.PrivateIpAddresses.1.PrivateIpAddress':
+ 'NetworkInterface.1.DeviceIndex': '2',
+ 'NetworkInterface.1.Description': 'description2',
+ 'NetworkInterface.1.DeleteOnTermination': 'false',
+ 'NetworkInterface.1.PrivateIpAddress': '10.0.1.54',
+ 'NetworkInterface.1.SubnetId': 'subnet_id2',
+ 'NetworkInterface.1.SecurityGroupId.0': 'group_id1',
+ 'NetworkInterface.1.SecurityGroupId.1': 'group_id2',
+ 'NetworkInterface.1.PrivateIpAddresses.0.Primary': 'false',
+ 'NetworkInterface.1.PrivateIpAddresses.0.PrivateIpAddress':
'10.0.1.10',
- 'NetworkInterface.2.PrivateIpAddresses.2.Primary': 'false',
- 'NetworkInterface.2.PrivateIpAddresses.2.PrivateIpAddress':
+ 'NetworkInterface.1.PrivateIpAddresses.1.Primary': 'false',
+ 'NetworkInterface.1.PrivateIpAddresses.1.PrivateIpAddress':
'10.0.1.11',
})
def test_add_prefix_to_serialization(self):
- return
collection = NetworkInterfaceCollection(self.network_interfaces_spec1,
self.network_interfaces_spec2)
params = {}
@@ -98,43 +109,92 @@ class TestNetworkInterfaceCollection(unittest.TestCase):
# we're just checking a few keys to make sure we get the proper
# prefix.
self.assertDictEqual(params, {
- 'LaunchSpecification.NetworkInterface.1.DeviceIndex': '1',
- 'LaunchSpecification.NetworkInterface.1.DeleteOnTermination':
+ 'LaunchSpecification.NetworkInterface.0.DeviceIndex': '1',
+ 'LaunchSpecification.NetworkInterface.0.DeleteOnTermination':
'false',
- 'LaunchSpecification.NetworkInterface.1.Description':
+ 'LaunchSpecification.NetworkInterface.0.Description':
'description1',
- 'LaunchSpecification.NetworkInterface.1.PrivateIpAddress':
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddress':
'10.0.0.54',
- 'LaunchSpecification.NetworkInterface.1.SubnetId': 'subnet_id',
- 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.1.Primary':
+ 'LaunchSpecification.NetworkInterface.0.SubnetId': 'subnet_id',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.0.Primary':
'false',
- 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.1.PrivateIpAddress':
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.0.PrivateIpAddress':
'10.0.0.10',
- 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.2.Primary': 'false',
- 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.2.PrivateIpAddress':
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.1.Primary': 'false',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.1.PrivateIpAddress':
'10.0.0.11',
- 'LaunchSpecification.NetworkInterface.2.DeviceIndex': '2',
- 'LaunchSpecification.NetworkInterface.2.Description':
+ 'LaunchSpecification.NetworkInterface.1.DeviceIndex': '2',
+ 'LaunchSpecification.NetworkInterface.1.Description':
'description2',
- 'LaunchSpecification.NetworkInterface.2.DeleteOnTermination':
+ 'LaunchSpecification.NetworkInterface.1.DeleteOnTermination':
'false',
- 'LaunchSpecification.NetworkInterface.2.PrivateIpAddress':
+ 'LaunchSpecification.NetworkInterface.1.PrivateIpAddress':
'10.0.1.54',
- 'LaunchSpecification.NetworkInterface.2.SubnetId': 'subnet_id2',
- 'LaunchSpecification.NetworkInterface.2.SecurityGroupId.1':
+ 'LaunchSpecification.NetworkInterface.1.SubnetId': 'subnet_id2',
+ 'LaunchSpecification.NetworkInterface.1.SecurityGroupId.0':
'group_id1',
- 'LaunchSpecification.NetworkInterface.2.SecurityGroupId.2':
+ 'LaunchSpecification.NetworkInterface.1.SecurityGroupId.1':
'group_id2',
- 'LaunchSpecification.NetworkInterface.2.PrivateIpAddresses.1.Primary':
+ 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.0.Primary':
'false',
- 'LaunchSpecification.NetworkInterface.2.PrivateIpAddresses.1.PrivateIpAddress':
+ 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.0.PrivateIpAddress':
'10.0.1.10',
- 'LaunchSpecification.NetworkInterface.2.PrivateIpAddresses.2.Primary':
+ 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.1.Primary':
'false',
- 'LaunchSpecification.NetworkInterface.2.PrivateIpAddresses.2.PrivateIpAddress':
+ 'LaunchSpecification.NetworkInterface.1.PrivateIpAddresses.1.PrivateIpAddress':
'10.0.1.11',
})
+ def test_cant_use_public_ip(self):
+ collection = NetworkInterfaceCollection(self.network_interfaces_spec3,
+ self.network_interfaces_spec1)
+ params = {}
+
+ # First, verify we can't incorrectly create multiple interfaces with
+ # on having a public IP.
+ with self.assertRaises(BotoClientError):
+ collection.build_list_params(params, prefix='LaunchSpecification.')
+
+ # Next, ensure it can't be on device index 1.
+ self.network_interfaces_spec3.device_index = 1
+ collection = NetworkInterfaceCollection(self.network_interfaces_spec3)
+ params = {}
+
+ with self.assertRaises(BotoClientError):
+ collection.build_list_params(params, prefix='LaunchSpecification.')
+
+ def test_public_ip(self):
+ # With public IP.
+ collection = NetworkInterfaceCollection(self.network_interfaces_spec3)
+ params = {}
+ collection.build_list_params(params, prefix='LaunchSpecification.')
+
+ self.assertDictEqual(params, {
+ 'LaunchSpecification.NetworkInterface.0.AssociatePublicIpAddress':
+ 'true',
+ 'LaunchSpecification.NetworkInterface.0.DeviceIndex': '0',
+ 'LaunchSpecification.NetworkInterface.0.DeleteOnTermination':
+ 'false',
+ 'LaunchSpecification.NetworkInterface.0.Description':
+ 'description2',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddress':
+ '10.0.1.54',
+ 'LaunchSpecification.NetworkInterface.0.SubnetId': 'subnet_id2',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.0.Primary':
+ 'false',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.0.PrivateIpAddress':
+ '10.0.1.10',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.1.Primary':
+ 'false',
+ 'LaunchSpecification.NetworkInterface.0.PrivateIpAddresses.1.PrivateIpAddress':
+ '10.0.1.11',
+ 'LaunchSpecification.NetworkInterface.0.SecurityGroupId.0':
+ 'group_id1',
+ 'LaunchSpecification.NetworkInterface.0.SecurityGroupId.1':
+ 'group_id2',
+ })
+
if __name__ == '__main__':
unittest.main()
diff --git a/tests/unit/ec2/test_securitygroup.py b/tests/unit/ec2/test_securitygroup.py
new file mode 100644
index 00000000..361dc256
--- /dev/null
+++ b/tests/unit/ec2/test_securitygroup.py
@@ -0,0 +1,212 @@
+#!/usr/bin/env python
+
+from tests.unit import unittest
+from tests.unit import AWSMockServiceTestCase
+
+import mock
+
+from boto.ec2.connection import EC2Connection
+from boto.ec2.securitygroup import SecurityGroup
+
+
+DESCRIBE_SECURITY_GROUP = r"""<?xml version="1.0" encoding="UTF-8"?>
+<DescribeSecurityGroupsResponse xmlns="http://ec2.amazonaws.com/doc/2013-06-15/">
+ <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
+ <securityGroupInfo>
+ <item>
+ <ownerId>111122223333</ownerId>
+ <groupId>sg-1a2b3c4d</groupId>
+ <groupName>WebServers</groupName>
+ <groupDescription>Web Servers</groupDescription>
+ <vpcId/>
+ <ipPermissions>
+ <item>
+ <ipProtocol>tcp</ipProtocol>
+ <fromPort>80</fromPort>
+ <toPort>80</toPort>
+ <groups/>
+ <ipRanges>
+ <item>
+ <cidrIp>0.0.0.0/0</cidrIp>
+ </item>
+ </ipRanges>
+ </item>
+ </ipPermissions>
+ <ipPermissionsEgress/>
+ </item>
+ <item>
+ <ownerId>111122223333</ownerId>
+ <groupId>sg-2a2b3c4d</groupId>
+ <groupName>RangedPortsBySource</groupName>
+ <groupDescription>Group A</groupDescription>
+ <ipPermissions>
+ <item>
+ <ipProtocol>tcp</ipProtocol>
+ <fromPort>6000</fromPort>
+ <toPort>7000</toPort>
+ <groups>
+ <item>
+ <userId>111122223333</userId>
+ <groupId>sg-3a2b3c4d</groupId>
+ <groupName>Group B</groupName>
+ </item>
+ </groups>
+ <ipRanges/>
+ </item>
+ </ipPermissions>
+ <ipPermissionsEgress/>
+ </item>
+ </securityGroupInfo>
+</DescribeSecurityGroupsResponse>"""
+
+DESCRIBE_INSTANCES = r"""<?xml version="1.0" encoding="UTF-8"?>
+<DescribeInstancesResponse xmlns="http://ec2.amazonaws.com/doc/2012-10-01/">
+ <requestId>c6132c74-b524-4884-87f5-0f4bde4a9760</requestId>
+ <reservationSet>
+ <item>
+ <reservationId>r-72ef4a0a</reservationId>
+ <ownerId>184906166255</ownerId>
+ <groupSet/>
+ <instancesSet>
+ <item>
+ <instanceId>i-instance</instanceId>
+ <imageId>ami-1624987f</imageId>
+ <instanceState>
+ <code>16</code>
+ <name>running</name>
+ </instanceState>
+ <privateDnsName/>
+ <dnsName/>
+ <reason/>
+ <keyName>mykeypair</keyName>
+ <amiLaunchIndex>0</amiLaunchIndex>
+ <productCodes/>
+ <instanceType>m1.small</instanceType>
+ <launchTime>2012-12-14T23:48:37.000Z</launchTime>
+ <placement>
+ <availabilityZone>us-east-1d</availabilityZone>
+ <groupName/>
+ <tenancy>default</tenancy>
+ </placement>
+ <kernelId>aki-88aa75e1</kernelId>
+ <monitoring>
+ <state>disabled</state>
+ </monitoring>
+ <subnetId>subnet-0dc60667</subnetId>
+ <vpcId>vpc-id</vpcId>
+ <privateIpAddress>10.0.0.67</privateIpAddress>
+ <sourceDestCheck>true</sourceDestCheck>
+ <groupSet>
+ <item>
+ <groupId>sg-1a2b3c4d</groupId>
+ <groupName>WebServerSG</groupName>
+ </item>
+ </groupSet>
+ <architecture>x86_64</architecture>
+ <rootDeviceType>ebs</rootDeviceType>
+ <rootDeviceName>/dev/sda1</rootDeviceName>
+ <blockDeviceMapping>
+ <item>
+ <deviceName>/dev/sda1</deviceName>
+ <ebs>
+ <volumeId>vol-id</volumeId>
+ <status>attached</status>
+ <attachTime>2012-12-14T23:48:43.000Z</attachTime>
+ <deleteOnTermination>true</deleteOnTermination>
+ </ebs>
+ </item>
+ </blockDeviceMapping>
+ <virtualizationType>paravirtual</virtualizationType>
+ <clientToken>foo</clientToken>
+ <tagSet>
+ <item>
+ <key>Name</key>
+ <value/>
+ </item>
+ </tagSet>
+ <hypervisor>xen</hypervisor>
+ <networkInterfaceSet>
+ <item>
+ <networkInterfaceId>eni-id</networkInterfaceId>
+ <subnetId>subnet-id</subnetId>
+ <vpcId>vpc-id</vpcId>
+ <description>Primary network interface</description>
+ <ownerId>ownerid</ownerId>
+ <status>in-use</status>
+ <privateIpAddress>10.0.0.67</privateIpAddress>
+ <sourceDestCheck>true</sourceDestCheck>
+ <groupSet>
+ <item>
+ <groupId>sg-id</groupId>
+ <groupName>WebServerSG</groupName>
+ </item>
+ </groupSet>
+ <attachment>
+ <attachmentId>eni-attach-id</attachmentId>
+ <deviceIndex>0</deviceIndex>
+ <status>attached</status>
+ <attachTime>2012-12-14T23:48:37.000Z</attachTime>
+ <deleteOnTermination>true</deleteOnTermination>
+ </attachment>
+ <privateIpAddressesSet>
+ <item>
+ <privateIpAddress>10.0.0.67</privateIpAddress>
+ <primary>true</primary>
+ </item>
+ <item>
+ <privateIpAddress>10.0.0.54</privateIpAddress>
+ <primary>false</primary>
+ </item>
+ <item>
+ <privateIpAddress>10.0.0.55</privateIpAddress>
+ <primary>false</primary>
+ </item>
+ </privateIpAddressesSet>
+ </item>
+ </networkInterfaceSet>
+ <ebsOptimized>false</ebsOptimized>
+ </item>
+ </instancesSet>
+ </item>
+ </reservationSet>
+</DescribeInstancesResponse>
+"""
+
+class TestDescribeSecurityGroups(AWSMockServiceTestCase):
+ connection_class = EC2Connection
+
+ def test_get_instances(self):
+ self.set_http_response(status_code=200, body=DESCRIBE_SECURITY_GROUP)
+ groups = self.service_connection.get_all_security_groups()
+
+ self.set_http_response(status_code=200, body=DESCRIBE_INSTANCES)
+ instances = groups[0].instances()
+
+ self.assertEqual(1, len(instances))
+ self.assertEqual(groups[0].id, instances[0].groups[0].id)
+
+
+class SecurityGroupTest(unittest.TestCase):
+ def test_add_rule(self):
+ sg = SecurityGroup()
+ self.assertEqual(len(sg.rules), 0)
+
+ # Regression: ``dry_run`` was being passed (but unhandled) before.
+ sg.add_rule(
+ ip_protocol='http',
+ from_port='80',
+ to_port='8080',
+ src_group_name='groupy',
+ src_group_owner_id='12345',
+ cidr_ip='10.0.0.1',
+ src_group_group_id='54321',
+ dry_run=False
+ )
+ self.assertEqual(len(sg.rules), 1)
+
+ def test_remove_rule_on_empty_group(self):
+ # Remove a rule from a group with no rules
+ sg = SecurityGroup()
+
+ with self.assertRaises(ValueError):
+ sg.remove_rule('ip', 80, 80, None, None, None, None)
diff --git a/tests/unit/ec2/test_volume.py b/tests/unit/ec2/test_volume.py
index fd2a4553..14f0bcb6 100644
--- a/tests/unit/ec2/test_volume.py
+++ b/tests/unit/ec2/test_volume.py
@@ -38,7 +38,12 @@ class VolumeTests(unittest.TestCase):
def test_startElement_calls_TaggedEC2Object_startElement_with_correct_args(self, startElement):
volume = Volume()
volume.startElement("some name", "some attrs", None)
- startElement.assert_called_with(volume, "some name", "some attrs", None)
+ startElement.assert_called_with(
+ volume,
+ "some name",
+ "some attrs",
+ None
+ )
@mock.patch("boto.ec2.volume.TaggedEC2Object.startElement")
def test_startElement_retval_not_None_returns_correct_thing(self, startElement):
@@ -120,43 +125,57 @@ class VolumeTests(unittest.TestCase):
def test_delete_calls_delete_volume(self):
self.volume_one.connection = mock.Mock()
self.volume_one.delete()
- self.volume_one.connection.delete_volume.assert_called_with(1)
+ self.volume_one.connection.delete_volume.assert_called_with(
+ 1,
+ dry_run=False
+ )
def test_attach_calls_attach_volume(self):
self.volume_one.connection = mock.Mock()
self.volume_one.attach("instance_id", "/dev/null")
- self.volume_one.connection.attach_volume.assert_called_with(1, "instance_id", "/dev/null")
+ self.volume_one.connection.attach_volume.assert_called_with(
+ 1,
+ "instance_id",
+ "/dev/null",
+ dry_run=False
+ )
def test_detach_calls_detach_volume(self):
self.volume_one.connection = mock.Mock()
self.volume_one.detach()
self.volume_one.connection.detach_volume.assert_called_with(
- 1, 2, "/dev/null", False)
+ 1, 2, "/dev/null", False, dry_run=False)
def test_detach_with_no_attach_data(self):
self.volume_two.connection = mock.Mock()
self.volume_two.detach()
self.volume_two.connection.detach_volume.assert_called_with(
- 1, None, None, False)
+ 1, None, None, False, dry_run=False)
def test_detach_with_force_calls_detach_volume_with_force(self):
self.volume_one.connection = mock.Mock()
self.volume_one.detach(True)
self.volume_one.connection.detach_volume.assert_called_with(
- 1, 2, "/dev/null", True)
+ 1, 2, "/dev/null", True, dry_run=False)
def test_create_snapshot_calls_connection_create_snapshot(self):
self.volume_one.connection = mock.Mock()
self.volume_one.create_snapshot()
self.volume_one.connection.create_snapshot.assert_called_with(
- 1, None)
+ 1,
+ None,
+ dry_run=False
+ )
def test_create_snapshot_with_description(self):
self.volume_one.connection = mock.Mock()
self.volume_one.create_snapshot("some description")
self.volume_one.connection.create_snapshot.assert_called_with(
- 1, "some description")
+ 1,
+ "some description",
+ dry_run=False
+ )
def test_volume_state_returns_status(self):
retval = self.volume_one.volume_state()
@@ -186,7 +205,7 @@ class VolumeTests(unittest.TestCase):
self.volume_one.connection.get_all_snapshots.return_value = []
self.volume_one.snapshots("owner", "restorable_by")
self.volume_one.connection.get_all_snapshots.assert_called_with(
- owner="owner", restorable_by="restorable_by")
+ owner="owner", restorable_by="restorable_by", dry_run=False)
class AttachmentSetTests(unittest.TestCase):
def check_that_attribute_has_been_set(self, name, value, attribute):
View Lorry Controller Status