| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You under the Apache License, Version 2.0 |
| # (the "License"); you may not use this file except in compliance with |
| # the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| |
| import os |
| import sys |
| import base64 |
| from datetime import datetime |
| from collections import OrderedDict |
| |
| from libcloud.test import MockHttp, LibcloudTestCase, unittest |
| from libcloud.utils.py3 import b, httplib, parse_qs |
| from libcloud.compute.base import ( |
| Node, |
| NodeSize, |
| NodeImage, |
| NodeLocation, |
| StorageVolume, |
| VolumeSnapshot, |
| ) |
| from libcloud.test.compute import TestCaseMixin |
| from libcloud.test.secrets import EC2_PARAMS |
| from libcloud.compute.types import ( |
| StorageVolumeState, |
| VolumeSnapshotState, |
| KeyPairDoesNotExistError, |
| ) |
| from libcloud.utils.iso8601 import UTC |
| from libcloud.test.file_fixtures import ComputeFileFixtures |
| from libcloud.compute.drivers.ec2 import ( |
| VALID_EC2_REGIONS, |
| EC2NodeDriver, |
| EucNodeDriver, |
| EC2NetworkSubnet, |
| NimbusNodeDriver, |
| EC2PlacementGroup, |
| IdempotentParamError, |
| ExEC2AvailabilityZone, |
| OutscaleSASNodeDriver, |
| ) |
| |
| null_fingerprint = "00:00:00:00:00:00:00:00:00:00:00:00:00:00:00:" + "00:00:00:00:00" |
| |
| |
| class BaseEC2Tests(LibcloudTestCase): |
| def test_instantiate_driver_valid_regions(self): |
| regions = VALID_EC2_REGIONS |
| regions = [d for d in regions if d != "nimbus" and d != "cn-north-1"] |
| |
| region_endpoints = [ |
| EC2NodeDriver(*EC2_PARAMS, **{"region": region}).connection.host for region in regions |
| ] |
| |
| # Verify that each driver doesn't get the same API host endpoint |
| self.assertEqual( |
| len(region_endpoints), |
| len(set(region_endpoints)), |
| "Multiple Region Drivers were given the same API endpoint", |
| ) |
| |
| def test_instantiate_driver_invalid_regions(self): |
| for region in ["invalid", "nimbus"]: |
| try: |
| EC2NodeDriver(*EC2_PARAMS, **{"region": region}) |
| except ValueError: |
| pass |
| else: |
| self.fail("Invalid region, but exception was not thrown") |
| |
| def test_list_sizes_valid_regions(self): |
| unsupported_regions = list() |
| |
| for region in VALID_EC2_REGIONS: |
| no_pricing = region in ["cn-north-1"] |
| driver = EC2NodeDriver(*EC2_PARAMS, **{"region": region}) |
| try: |
| sizes = driver.list_sizes() |
| if no_pricing: |
| self.assertTrue(all([s.price is None for s in sizes])) |
| except Exception: |
| unsupported_regions.append(region) |
| |
| if unsupported_regions: |
| self.fail("Cannot list sizes from ec2 regions: %s" % unsupported_regions) |
| |
| |
| class EC2Tests(LibcloudTestCase, TestCaseMixin): |
| image_name = "ec2-public-images/fedora-8-i386-base-v1.04.manifest.xml" |
| region = "us-east-1" |
| |
| def setUp(self): |
| EC2MockHttp.test = self |
| EC2NodeDriver.connectionCls.conn_class = EC2MockHttp |
| EC2MockHttp.use_param = "Action" |
| EC2MockHttp.type = None |
| |
| self.driver = EC2NodeDriver(*EC2_PARAMS, **{"region": self.region}) |
| |
| def test_regions_and_signature_versions(self): |
| # Verify that correct signature versions are used for each region |
| driver = EC2NodeDriver(*EC2_PARAMS, region="us-east-1") |
| self.assertEqual(driver.signature_version, "2") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="us-east-2") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="eu-west-1") |
| self.assertEqual(driver.signature_version, "2") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="eu-west-3") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="ca-central-1") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="eu-central-1") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="eu-south-1") |
| self.assertEqual(driver.signature_version, "4") |
| |
| # Verify that signature_version can be overriden via constructor argument |
| driver = EC2NodeDriver(*EC2_PARAMS, region="us-east-1", signature_version="2") |
| self.assertEqual(driver.signature_version, "2") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="us-east-1", signature_version="4") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="eu-central-1", signature_version="2") |
| self.assertEqual(driver.signature_version, "2") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="eu-central-1", signature_version="4") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="ap-east-1") |
| self.assertEqual(driver.signature_version, "2") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="ap-east-1", signature_version="4") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="af-south-1") |
| self.assertEqual(driver.signature_version, "4") |
| |
| driver = EC2NodeDriver(*EC2_PARAMS, region="af-south-1", signature_version="4") |
| self.assertEqual(driver.signature_version, "4") |
| |
| def test_instantiate_driver_with_token(self): |
| token = "temporary_credentials_token" |
| driver = EC2NodeDriver(*EC2_PARAMS, **{"region": self.region, "token": token}) |
| self.assertTrue(hasattr(driver, "token"), "Driver has no attribute token") |
| self.assertEqual(token, driver.token, "Driver token does not match with provided token") |
| |
| def test_driver_with_token_signature_version(self): |
| token = "temporary_credentials_token" |
| driver = EC2NodeDriver(*EC2_PARAMS, **{"region": self.region, "token": token}) |
| kwargs = driver._ex_connection_class_kwargs() |
| self.assertIn("signature_version", kwargs) |
| self.assertEqual( |
| "4", |
| kwargs["signature_version"], |
| "Signature version is not 4 with temporary credentials", |
| ) |
| |
| def test_create_node(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| node = self.driver.create_node(name="foo", image=image, size=size) |
| self.assertEqual(node.id, "i-2ba64342") |
| self.assertEqual(node.name, "foo") |
| self.assertEqual(node.extra["tags"]["Name"], "foo") |
| self.assertEqual(len(node.extra["tags"]), 1) |
| |
| def test_create_node_with_ex_mincount(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| node = self.driver.create_node( |
| name="foo", image=image, size=size, ex_mincount=1, ex_maxcount=10 |
| ) |
| self.assertEqual(node.id, "i-2ba64342") |
| self.assertEqual(node.name, "foo") |
| self.assertEqual(node.extra["tags"]["Name"], "foo") |
| self.assertEqual(len(node.extra["tags"]), 1) |
| |
| def test_create_node_with_ex_assign_public_ip(self): |
| # assertions are done in _create_ex_assign_public_ip_RunInstances |
| EC2MockHttp.type = "create_ex_assign_public_ip" |
| image = NodeImage(id="ami-11111111", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| subnet = EC2NetworkSubnet("subnet-11111111", "test_subnet", "pending") |
| self.driver.create_node( |
| name="foo", |
| image=image, |
| size=size, |
| ex_subnet=subnet, |
| ex_security_group_ids=["sg-11111111"], |
| ex_assign_public_ip=True, |
| ) |
| |
| def test_create_node_with_ex_terminate_on_shutdown(self): |
| EC2MockHttp.type = "create_ex_terminate_on_shutdown" |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| # The important part about the test is asserted inside |
| # EC2MockHttp._create_ex_terminate_on_shutdown |
| self.driver.create_node(name="foo", image=image, size=size, ex_terminate_on_shutdown=True) |
| |
| def test_create_node_with_metadata(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| node = self.driver.create_node( |
| name="foo", image=image, size=size, ex_metadata={"Bar": "baz", "Num": "42"} |
| ) |
| self.assertEqual(node.name, "foo") |
| self.assertEqual(node.extra["tags"]["Name"], "foo") |
| self.assertEqual(node.extra["tags"]["Bar"], "baz") |
| self.assertEqual(node.extra["tags"]["Num"], "42") |
| self.assertEqual(len(node.extra["tags"]), 3) |
| |
| def test_create_node_idempotent(self): |
| EC2MockHttp.type = "idempotent" |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| token = "testclienttoken" |
| node = self.driver.create_node(name="foo", image=image, size=size, ex_clienttoken=token) |
| self.assertEqual(node.id, "i-2ba64342") |
| self.assertEqual(node.extra["client_token"], token) |
| |
| # from: http://docs.amazonwebservices.com/AWSEC2/latest/DeveloperGuide/index.html?Run_Instance_Idempotency.html |
| # If you repeat the request with the same client token, but change |
| # another request parameter, Amazon EC2 returns an |
| # IdempotentParameterMismatch error. |
| # In our case, changing the parameter doesn't actually matter since we |
| # are forcing the error response fixture. |
| EC2MockHttp.type = "idempotent_mismatch" |
| |
| idem_error = None |
| # different count |
| try: |
| self.driver.create_node( |
| name="foo", |
| image=image, |
| size=size, |
| ex_mincount="2", |
| ex_maxcount="2", |
| ex_clienttoken=token, |
| ) |
| except IdempotentParamError as e: |
| idem_error = e |
| self.assertTrue(idem_error is not None) |
| |
| def test_create_node_no_availability_zone(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| node = self.driver.create_node(name="foo", image=image, size=size) |
| location = NodeLocation(0, "Amazon US N. Virginia", "US", self.driver) |
| self.assertEqual(node.id, "i-2ba64342") |
| node = self.driver.create_node(name="foo", image=image, size=size, location=location) |
| self.assertEqual(node.id, "i-2ba64342") |
| self.assertEqual(node.name, "foo") |
| |
| def test_list_nodes(self): |
| node = self.driver.list_nodes()[0] |
| public_ips = sorted(node.public_ips) |
| self.assertEqual(node.id, "i-4382922a") |
| self.assertEqual(node.name, node.id) |
| self.assertEqual(len(node.public_ips), 2) |
| |
| self.assertEqual(node.extra["launch_time"], "2013-12-02T11:58:11.000Z") |
| self.assertEqual(node.created_at, datetime(2013, 12, 2, 11, 58, 11, tzinfo=UTC)) |
| |
| self.assertTrue("instance_type" in node.extra) |
| self.assertEqual(node.extra["availability"], "us-east-1d") |
| self.assertEqual(node.extra["key_name"], "fauxkey") |
| self.assertEqual(node.extra["monitoring"], "disabled") |
| self.assertEqual(node.extra["image_id"], "ami-3215fe5a") |
| self.assertEqual(len(node.extra["groups"]), 2) |
| self.assertEqual(len(node.extra["block_device_mapping"]), 1) |
| self.assertEqual(node.extra["block_device_mapping"][0]["device_name"], "/dev/sda1") |
| self.assertEqual(node.extra["block_device_mapping"][0]["ebs"]["volume_id"], "vol-5e312311") |
| self.assertTrue(node.extra["block_device_mapping"][0]["ebs"]["delete"]) |
| self.assertEqual(node.extra["block_device_mapping"][0]["ebs"]["status"], "attached") |
| self.assertEqual( |
| node.extra["block_device_mapping"][0]["ebs"]["attach_time"], |
| datetime(2013, 4, 9, 18, 1, 1, tzinfo=UTC), |
| ) |
| self.assertEqual(public_ips[0], "1.2.3.4") |
| |
| nodes = self.driver.list_nodes(ex_node_ids=["i-4382922a", "i-8474834a"]) |
| ret_node1 = nodes[0] |
| ret_node2 = nodes[1] |
| |
| self.assertEqual(ret_node1.id, "i-4382922a") |
| self.assertEqual(ret_node2.id, "i-8474834a") |
| self.assertEqual(ret_node2.name, "Test Server 2") |
| self.assertEqual(ret_node2.extra["subnet_id"], "subnet-5fd9d412") |
| self.assertEqual(ret_node2.extra["vpc_id"], "vpc-61dcd30e") |
| self.assertEqual(ret_node2.extra["tags"]["Group"], "VPC Test") |
| |
| self.assertEqual(ret_node1.extra["launch_time"], "2013-12-02T11:58:11.000Z") |
| self.assertEqual(ret_node1.created_at, datetime(2013, 12, 2, 11, 58, 11, tzinfo=UTC)) |
| self.assertEqual(ret_node2.extra["launch_time"], "2013-12-02T15:58:29.000Z") |
| self.assertEqual(ret_node2.created_at, datetime(2013, 12, 2, 15, 58, 29, tzinfo=UTC)) |
| |
| self.assertIn("instance_type", ret_node1.extra) |
| self.assertIn("instance_type", ret_node2.extra) |
| |
| def test_ex_list_reserved_nodes(self): |
| node = self.driver.ex_list_reserved_nodes()[0] |
| self.assertEqual(node.id, "93bbbca2-c500-49d0-9ede-9d8737400498") |
| self.assertEqual(node.state, "active") |
| self.assertEqual(node.extra["instance_type"], "t1.micro") |
| self.assertEqual(node.extra["availability"], "us-east-1b") |
| self.assertEqual(node.extra["start"], "2013-06-18T12:07:53.161Z") |
| self.assertEqual(node.extra["end"], "2014-06-18T12:07:53.161Z") |
| self.assertEqual(node.extra["duration"], 31536000) |
| self.assertEqual(node.extra["usage_price"], 0.012) |
| self.assertEqual(node.extra["fixed_price"], 23.0) |
| self.assertEqual(node.extra["instance_count"], 1) |
| self.assertEqual(node.extra["description"], "Linux/UNIX") |
| self.assertEqual(node.extra["instance_tenancy"], "default") |
| self.assertEqual(node.extra["currency_code"], "USD") |
| self.assertEqual(node.extra["offering_type"], "Light Utilization") |
| |
| def test_list_location(self): |
| locations = self.driver.list_locations() |
| self.assertTrue(len(locations) > 0) |
| self.assertEqual(locations[0].name, "eu-west-1a") |
| self.assertTrue(locations[0].availability_zone is not None) |
| self.assertTrue(isinstance(locations[0].availability_zone, ExEC2AvailabilityZone)) |
| |
| def test_list_security_groups(self): |
| groups = self.driver.ex_list_security_groups() |
| self.assertEqual(groups, ["WebServers", "RangedPortsBySource"]) |
| |
| def test_ex_delete_security_group_by_id(self): |
| group_id = "sg-443d0a12" |
| retValue = self.driver.ex_delete_security_group_by_id(group_id) |
| self.assertTrue(retValue) |
| |
| def test_delete_security_group_by_name(self): |
| group_name = "WebServers" |
| retValue = self.driver.ex_delete_security_group_by_name(group_name) |
| self.assertTrue(retValue) |
| |
| def test_ex_delete_security_group(self): |
| name = "WebServers" |
| retValue = self.driver.ex_delete_security_group(name) |
| self.assertTrue(retValue) |
| |
| def test_authorize_security_group(self): |
| resp = self.driver.ex_authorize_security_group("TestGroup", "22", "22", "0.0.0.0/0") |
| self.assertTrue(resp) |
| |
| def test_authorize_security_group_ingress(self): |
| ranges = ["1.1.1.1/32", "2.2.2.2/32"] |
| description = "automated authorised IP ingress test" |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, cidr_ips=ranges, description=description |
| ) |
| self.assertTrue(resp) |
| groups = [{"group_id": "sg-949265ff"}] |
| description = "automated authorised group ingress test" |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 23, group_pairs=groups, description=description |
| ) |
| self.assertTrue(resp) |
| |
| def test_authorize_security_group_egress(self): |
| ranges = ["1.1.1.1/32", "2.2.2.2/32"] |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, cidr_ips=ranges |
| ) |
| self.assertTrue(resp) |
| groups = [{"group_id": "sg-949265ff"}] |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, group_pairs=groups |
| ) |
| self.assertTrue(resp) |
| |
| def test_revoke_security_group_ingress(self): |
| ranges = ["1.1.1.1/32", "2.2.2.2/32"] |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, cidr_ips=ranges |
| ) |
| self.assertTrue(resp) |
| groups = [{"group_id": "sg-949265ff"}] |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, group_pairs=groups |
| ) |
| self.assertTrue(resp) |
| |
| def test_revoke_security_group_egress(self): |
| ranges = ["1.1.1.1/32", "2.2.2.2/32"] |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, cidr_ips=ranges |
| ) |
| self.assertTrue(resp) |
| groups = [{"group_id": "sg-949265ff"}] |
| resp = self.driver.ex_authorize_security_group_ingress( |
| "sg-42916629", 22, 22, group_pairs=groups |
| ) |
| self.assertTrue(resp) |
| |
| def test_reboot_node(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| ret = self.driver.reboot_node(node) |
| self.assertTrue(ret) |
| |
| def test_ex_start_node(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| ret = self.driver.ex_start_node(node) |
| self.assertTrue(ret) |
| |
| def test_ex_stop_node(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| ret = self.driver.ex_stop_node(node) |
| self.assertTrue(ret) |
| |
| def test_ex_create_node_with_ex_blockdevicemappings(self): |
| EC2MockHttp.type = "create_ex_blockdevicemappings" |
| |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| mappings = [ |
| {"DeviceName": "/dev/sda1", "Ebs.VolumeSize": 10}, |
| {"DeviceName": "/dev/sdb", "VirtualName": "ephemeral0"}, |
| {"DeviceName": "/dev/sdc", "VirtualName": "ephemeral1"}, |
| ] |
| node = self.driver.create_node( |
| name="foo", image=image, size=size, ex_blockdevicemappings=mappings |
| ) |
| self.assertEqual(node.id, "i-2ba64342") |
| |
| def test_ex_create_node_with_ex_blockdevicemappings_attribute_error(self): |
| EC2MockHttp.type = "create_ex_blockdevicemappings" |
| |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| mappings = "this should be a list" |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_node, |
| name="foo", |
| image=image, |
| size=size, |
| ex_blockdevicemappings=mappings, |
| ) |
| |
| mappings = ["this should be a dict"] |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_node, |
| name="foo", |
| image=image, |
| size=size, |
| ex_blockdevicemappings=mappings, |
| ) |
| |
| def test_destroy_node(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| ret = self.driver.destroy_node(node) |
| self.assertTrue(ret) |
| |
| def test_list_sizes(self): |
| region_old = self.driver.region_name |
| |
| names = [ |
| ("ec2_us_east", "us-east-1"), |
| ("ec2_us_west", "us-west-1"), |
| ("ec2_us_west", "us-west-2"), |
| ("ec2_eu_west", "eu-west-1"), |
| ("ec2_ap_southeast", "ap-southeast-1"), |
| ("ec2_ap_northeast", "ap-northeast-1"), |
| ("ec2_ap_southeast_2", "ap-southeast-2"), |
| ("ec2_ap_south_1", "ap-south-1"), |
| ] |
| |
| for api_name, region_name in names: |
| self.driver.api_name = api_name |
| self.driver.region_name = region_name |
| sizes = self.driver.list_sizes() |
| self.assertNotEqual(len(sizes), 0) |
| self.driver.region_name = region_old |
| |
| def test_ex_create_node_with_ex_iam_profile(self): |
| iamProfile = { |
| "id": "AIDGPMS9RO4H3FEXAMPLE", |
| "name": "Foo", |
| "arn": "arn:aws:iam:...", |
| } |
| |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| EC2MockHttp.type = None |
| node1 = self.driver.create_node(name="foo", image=image, size=size) |
| EC2MockHttp.type = "ex_iam_profile" |
| node2 = self.driver.create_node( |
| name="bar", image=image, size=size, ex_iamprofile=iamProfile["name"] |
| ) |
| node3 = self.driver.create_node( |
| name="bar", image=image, size=size, ex_iamprofile=iamProfile["arn"] |
| ) |
| |
| self.assertFalse(node1.extra["iam_profile"]) |
| self.assertEqual(node2.extra["iam_profile"], iamProfile["id"]) |
| self.assertEqual(node3.extra["iam_profile"], iamProfile["id"]) |
| |
| def test_ex_create_node_with_ex_spot(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| EC2MockHttp.type = "ex_spot" |
| node = self.driver.create_node(name="foo", image=image, size=size, ex_spot=True) |
| self.assertEqual(node.extra["instance_lifecycle"], "spot") |
| node = self.driver.create_node( |
| name="foo", image=image, size=size, ex_spot=True, ex_spot_max_price=1.5 |
| ) |
| self.assertEqual(node.extra["instance_lifecycle"], "spot") |
| |
| def test_list_images(self): |
| images = self.driver.list_images() |
| |
| self.assertEqual(len(images), 3) |
| location = "123456788908/Test Image" |
| self.assertEqual(images[0].id, "ami-57ba933a") |
| self.assertEqual(images[0].name, "Test Image") |
| self.assertEqual(images[0].extra["image_location"], location) |
| self.assertEqual(images[0].extra["architecture"], "x86_64") |
| self.assertEqual(len(images[0].extra["block_device_mapping"]), 2) |
| ephemeral = images[0].extra["block_device_mapping"][1]["virtual_name"] |
| self.assertEqual(ephemeral, "ephemeral0") |
| billing_product1 = images[0].extra["billing_products"][0] |
| self.assertEqual(billing_product1, "ab-5dh78019") |
| self.assertIsInstance(images[0].extra["creation_date"], datetime) |
| |
| location = "123456788908/Test Image 2" |
| self.assertEqual(images[1].id, "ami-85b2a8ae") |
| self.assertEqual(images[1].name, "Test Image 2") |
| self.assertEqual(images[1].extra["image_location"], location) |
| self.assertEqual(images[1].extra["architecture"], "x86_64") |
| size = images[1].extra["block_device_mapping"][0]["ebs"]["volume_size"] |
| billing_product2 = images[1].extra["billing_products"][0] |
| self.assertEqual(billing_product2, "as-6dr90319") |
| self.assertEqual(size, 20) |
| self.assertIsNone(images[1].extra["creation_date"]) |
| |
| self.assertEqual(images[2].id, "ami-85b2a8ac") |
| self.assertEqual(images[2].name, "Test Image 3") |
| self.assertEqual(size, 20) |
| self.assertIsNone(images[2].extra["creation_date"]) |
| |
| def test_list_images_with_image_ids(self): |
| EC2MockHttp.type = "ex_imageids" |
| images = self.driver.list_images(ex_image_ids=["ami-57ba933a"]) |
| |
| self.assertEqual(len(images), 1) |
| self.assertEqual(images[0].name, "Test Image") |
| |
| def test_list_images_with_executable_by(self): |
| images = self.driver.list_images(ex_executableby="self") |
| |
| self.assertEqual(len(images), 3) |
| |
| def test_get_image(self): |
| image = self.driver.get_image("ami-57ba933a") |
| self.assertEqual(image.id, "ami-57ba933a") |
| self.assertEqual(image.name, "Test Image") |
| self.assertEqual(image.extra["architecture"], "x86_64") |
| self.assertEqual(len(image.extra["block_device_mapping"]), 2) |
| self.assertEqual(image.extra["billing_products"][0], "ab-5dh78019") |
| self.assertIsInstance(image.extra["creation_date"], datetime) |
| |
| def test_copy_image(self): |
| image = self.driver.list_images()[0] |
| resp = self.driver.copy_image( |
| image, "us-east-1", name="Faux Image", description="Test Image Copy" |
| ) |
| self.assertEqual(resp.id, "ami-4db38224") |
| |
| def test_create_image(self): |
| node = self.driver.list_nodes()[0] |
| |
| mapping = [ |
| { |
| "VirtualName": None, |
| "Ebs": { |
| "VolumeSize": 10, |
| "VolumeType": "standard", |
| "DeleteOnTermination": "true", |
| }, |
| "DeviceName": "/dev/sda1", |
| } |
| ] |
| |
| resp = self.driver.create_image( |
| node, "New Image", description="New EBS Image", block_device_mapping=mapping |
| ) |
| self.assertEqual(resp.id, "ami-e9b38280") |
| |
| def test_create_image_no_mapping(self): |
| node = self.driver.list_nodes()[0] |
| |
| resp = self.driver.create_image(node, "New Image", description="New EBS Image") |
| self.assertEqual(resp.id, "ami-e9b38280") |
| |
| def delete_image(self): |
| images = self.driver.list_images() |
| image = images[0] |
| |
| resp = self.driver.delete_image(image) |
| self.assertTrue(resp) |
| |
| def ex_register_image(self): |
| mapping = [{"DeviceName": "/dev/sda1", "Ebs": {"SnapshotId": "snap-5ade3e4e"}}] |
| image = self.driver.ex_register_image( |
| name="Test Image", |
| root_device_name="/dev/sda1", |
| description="My Image", |
| architecture="x86_64", |
| block_device_mapping=mapping, |
| ena_support=True, |
| billing_products=["ab-5dh78019"], |
| sriov_net_support="simple", |
| ) |
| self.assertEqual(image.id, "ami-57c2fb3e") |
| |
| def test_ex_import_snapshot(self): |
| disk_container = [ |
| { |
| "Description": "Dummy import snapshot task", |
| "Format": "raw", |
| "UserBucket": {"S3Bucket": "dummy-bucket", "S3Key": "dummy-key"}, |
| } |
| ] |
| |
| snap = self.driver.ex_import_snapshot(disk_container=disk_container) |
| self.assertEqual(snap.id, "snap-0ea83e8a87e138f39") |
| |
| def test_wait_for_import_snapshot_completion(self): |
| snap = self.driver._wait_for_import_snapshot_completion( |
| import_task_id="import-snap-fhdysyq6" |
| ) |
| self.assertEqual(snap.id, "snap-0ea83e8a87e138f39") |
| |
| def test_timeout_wait_for_import_snapshot_completion(self): |
| import_task_id = "import-snap-fhdysyq6" |
| EC2MockHttp.type = "timeout" |
| with self.assertRaises(Exception) as context: |
| self.driver._wait_for_import_snapshot_completion( |
| import_task_id=import_task_id, timeout=0.01, interval=0.001 |
| ) |
| self.assertEqual( |
| "Timeout while waiting for import task Id %s" % import_task_id, |
| str(context.exception), |
| ) |
| |
| def test_ex_describe_import_snapshot_tasks(self): |
| snap = self.driver.ex_describe_import_snapshot_tasks(import_task_id="import-snap-fh7y6i6w<") |
| |
| self.assertEqual(snap.snapshotId, "snap-0ea83e8a87e138f39") |
| self.assertEqual(snap.status, "completed") |
| |
| def test_ex_list_availability_zones(self): |
| availability_zones = self.driver.ex_list_availability_zones() |
| availability_zone = availability_zones[0] |
| self.assertTrue(len(availability_zones) > 0) |
| self.assertEqual(availability_zone.name, "eu-west-1a") |
| self.assertEqual(availability_zone.zone_state, "available") |
| self.assertEqual(availability_zone.region_name, "eu-west-1") |
| |
| def test_list_keypairs(self): |
| keypairs = self.driver.list_key_pairs() |
| |
| self.assertEqual(len(keypairs), 1) |
| self.assertEqual(keypairs[0].name, "gsg-keypair") |
| self.assertEqual(keypairs[0].fingerprint, null_fingerprint) |
| |
| # Test old deprecated method |
| keypairs = self.driver.ex_list_keypairs() |
| |
| self.assertEqual(len(keypairs), 1) |
| self.assertEqual(keypairs[0]["keyName"], "gsg-keypair") |
| self.assertEqual(keypairs[0]["keyFingerprint"], null_fingerprint) |
| |
| def test_get_key_pair(self): |
| EC2MockHttp.type = "get_one" |
| |
| key_pair = self.driver.get_key_pair(name="gsg-keypair") |
| self.assertEqual(key_pair.name, "gsg-keypair") |
| |
| def test_get_key_pair_does_not_exist(self): |
| EC2MockHttp.type = "doesnt_exist" |
| |
| self.assertRaises(KeyPairDoesNotExistError, self.driver.get_key_pair, name="test-key-pair") |
| |
| def test_create_key_pair(self): |
| key_pair = self.driver.create_key_pair(name="test-keypair") |
| |
| fingerprint = "1f:51:ae:28:bf:89:e9:d8:1f:25:5d" ":37:2d:7d:b8:ca:9f:f5:f1:6f" |
| |
| self.assertEqual(key_pair.name, "my-key-pair") |
| self.assertEqual(key_pair.fingerprint, fingerprint) |
| self.assertTrue(key_pair.private_key is not None) |
| |
| # Test old and deprecated method |
| key_pair = self.driver.ex_create_keypair(name="test-keypair") |
| self.assertEqual(key_pair["keyFingerprint"], fingerprint) |
| self.assertTrue(key_pair["keyMaterial"] is not None) |
| |
| def test_ex_describe_all_keypairs(self): |
| keys = self.driver.ex_describe_all_keypairs() |
| self.assertEqual(keys, ["gsg-keypair"]) |
| |
| def test_list_key_pairs(self): |
| keypair1 = self.driver.list_key_pairs()[0] |
| |
| self.assertEqual(keypair1.name, "gsg-keypair") |
| self.assertEqual(keypair1.fingerprint, null_fingerprint) |
| |
| # Test backward compatibility |
| keypair2 = self.driver.ex_describe_keypairs("gsg-keypair") |
| |
| self.assertEqual(keypair2["keyName"], "gsg-keypair") |
| self.assertEqual(keypair2["keyFingerprint"], null_fingerprint) |
| |
| def test_delete_key_pair(self): |
| keypair = self.driver.list_key_pairs()[0] |
| success = self.driver.delete_key_pair(keypair) |
| |
| self.assertTrue(success) |
| |
| # Test old and deprecated method |
| resp = self.driver.ex_delete_keypair("gsg-keypair") |
| self.assertTrue(resp) |
| |
| def test_ex_describe_tags(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| tags = self.driver.ex_describe_tags(resource=node) |
| |
| self.assertEqual(len(tags), 3) |
| self.assertTrue("tag" in tags) |
| self.assertTrue("owner" in tags) |
| self.assertTrue("stack" in tags) |
| |
| def test_import_key_pair_from_string(self): |
| path = os.path.join(os.path.dirname(__file__), "fixtures", "misc", "test_rsa.pub") |
| |
| with open(path) as fp: |
| key_material = fp.read() |
| |
| key = self.driver.import_key_pair_from_string(name="keypair", key_material=key_material) |
| self.assertEqual(key.name, "keypair") |
| self.assertEqual(key.fingerprint, null_fingerprint) |
| |
| # Test old and deprecated method |
| key = self.driver.ex_import_keypair_from_string("keypair", key_material) |
| self.assertEqual(key["keyName"], "keypair") |
| self.assertEqual(key["keyFingerprint"], null_fingerprint) |
| |
| def test_import_key_pair_from_file(self): |
| path = os.path.join(os.path.dirname(__file__), "fixtures", "misc", "test_rsa.pub") |
| |
| key = self.driver.import_key_pair_from_file("keypair", path) |
| self.assertEqual(key.name, "keypair") |
| self.assertEqual(key.fingerprint, null_fingerprint) |
| |
| # Test old and deprecated method |
| key = self.driver.ex_import_keypair("keypair", path) |
| self.assertEqual(key["keyName"], "keypair") |
| self.assertEqual(key["keyFingerprint"], null_fingerprint) |
| |
| def test_ex_create_tags(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| self.driver.ex_create_tags(node, {"sample": "tag"}) |
| |
| def test_ex_delete_tags(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| self.driver.ex_delete_tags(node, {"sample": "tag"}) |
| |
| def test_ex_delete_tags2(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| self.driver.ex_create_tags(node, {"sample": "another tag"}) |
| self.driver.ex_delete_tags(node, {"sample": None}) |
| |
| def test_ex_describe_addresses_for_node(self): |
| node1 = Node("i-4382922a", None, None, None, None, self.driver) |
| ip_addresses1 = self.driver.ex_describe_addresses_for_node(node1) |
| node2 = Node("i-4382922b", None, None, None, None, self.driver) |
| ip_addresses2 = sorted(self.driver.ex_describe_addresses_for_node(node2)) |
| node3 = Node("i-4382922g", None, None, None, None, self.driver) |
| ip_addresses3 = sorted(self.driver.ex_describe_addresses_for_node(node3)) |
| |
| self.assertEqual(len(ip_addresses1), 1) |
| self.assertEqual(ip_addresses1[0], "1.2.3.4") |
| |
| self.assertEqual(len(ip_addresses2), 2) |
| self.assertEqual(ip_addresses2[0], "1.2.3.5") |
| self.assertEqual(ip_addresses2[1], "1.2.3.6") |
| |
| self.assertEqual(len(ip_addresses3), 0) |
| |
| def test_ex_describe_addresses(self): |
| node1 = Node("i-4382922a", None, None, None, None, self.driver) |
| node2 = Node("i-4382922g", None, None, None, None, self.driver) |
| nodes_elastic_ips1 = self.driver.ex_describe_addresses([node1]) |
| nodes_elastic_ips2 = self.driver.ex_describe_addresses([node2]) |
| |
| self.assertEqual(len(nodes_elastic_ips1), 1) |
| self.assertTrue(node1.id in nodes_elastic_ips1) |
| self.assertEqual(nodes_elastic_ips1[node1.id], ["1.2.3.4"]) |
| |
| self.assertEqual(len(nodes_elastic_ips2), 1) |
| self.assertTrue(node2.id in nodes_elastic_ips2) |
| self.assertEqual(nodes_elastic_ips2[node2.id], []) |
| |
| def test_ex_describe_all_addresses(self): |
| EC2MockHttp.type = "all_addresses" |
| elastic_ips1 = self.driver.ex_describe_all_addresses() |
| elastic_ips2 = self.driver.ex_describe_all_addresses(only_associated=True) |
| self.assertEqual("1.2.3.7", elastic_ips1[3].ip) |
| self.assertEqual("vpc", elastic_ips1[3].domain) |
| self.assertEqual("eipalloc-992a5cf8", elastic_ips1[3].extra["allocation_id"]) |
| |
| self.assertEqual(len(elastic_ips2), 2) |
| self.assertEqual("1.2.3.5", elastic_ips2[1].ip) |
| self.assertEqual("vpc", elastic_ips2[1].domain) |
| |
| def test_ex_allocate_address(self): |
| elastic_ip = self.driver.ex_allocate_address() |
| self.assertEqual("192.0.2.1", elastic_ip.ip) |
| self.assertEqual("standard", elastic_ip.domain) |
| EC2MockHttp.type = "vpc" |
| elastic_ip = self.driver.ex_allocate_address(domain="vpc") |
| self.assertEqual("192.0.2.2", elastic_ip.ip) |
| self.assertEqual("vpc", elastic_ip.domain) |
| self.assertEqual("eipalloc-666d7f04", elastic_ip.extra["allocation_id"]) |
| |
| def test_ex_release_address(self): |
| EC2MockHttp.type = "all_addresses" |
| elastic_ips = self.driver.ex_describe_all_addresses() |
| EC2MockHttp.type = "" |
| ret = self.driver.ex_release_address(elastic_ips[2]) |
| self.assertTrue(ret) |
| ret = self.driver.ex_release_address(elastic_ips[0], domain="vpc") |
| self.assertTrue(ret) |
| self.assertRaises( |
| AttributeError, |
| self.driver.ex_release_address, |
| elastic_ips[0], |
| domain="bogus", |
| ) |
| |
| def test_ex_associate_address_with_node(self): |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| EC2MockHttp.type = "all_addresses" |
| elastic_ips = self.driver.ex_describe_all_addresses() |
| EC2MockHttp.type = "" |
| ret1 = self.driver.ex_associate_address_with_node(node, elastic_ips[2]) |
| ret2 = self.driver.ex_associate_addresses(node, elastic_ips[2]) |
| self.assertEqual(None, ret1) |
| self.assertEqual(None, ret2) |
| EC2MockHttp.type = "vpc" |
| ret3 = self.driver.ex_associate_address_with_node(node, elastic_ips[3], domain="vpc") |
| ret4 = self.driver.ex_associate_addresses(node, elastic_ips[3], domain="vpc") |
| self.assertEqual("eipassoc-167a8073", ret3) |
| self.assertEqual("eipassoc-167a8073", ret4) |
| self.assertRaises( |
| AttributeError, |
| self.driver.ex_associate_address_with_node, |
| node, |
| elastic_ips[1], |
| domain="bogus", |
| ) |
| |
| def test_ex_disassociate_address(self): |
| EC2MockHttp.type = "all_addresses" |
| elastic_ips = self.driver.ex_describe_all_addresses() |
| EC2MockHttp.type = "" |
| ret = self.driver.ex_disassociate_address(elastic_ips[2]) |
| self.assertTrue(ret) |
| # Test a VPC disassociation |
| ret = self.driver.ex_disassociate_address(elastic_ips[1], domain="vpc") |
| self.assertTrue(ret) |
| self.assertRaises( |
| AttributeError, |
| self.driver.ex_disassociate_address, |
| elastic_ips[1], |
| domain="bogus", |
| ) |
| |
| def test_ex_change_node_size_same_size(self): |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| node = Node( |
| "i-4382922a", |
| None, |
| None, |
| None, |
| None, |
| self.driver, |
| extra={"instancetype": "m1.small"}, |
| ) |
| |
| try: |
| self.driver.ex_change_node_size(node=node, new_size=size) |
| except ValueError: |
| pass |
| else: |
| self.fail("Same size was passed, but an exception was not thrown") |
| |
| def test_ex_change_node_size(self): |
| size = NodeSize("m1.large", "Small Instance", None, None, None, None, driver=self.driver) |
| node = Node( |
| "i-4382922a", |
| None, |
| None, |
| None, |
| None, |
| self.driver, |
| extra={"instancetype": "m1.small"}, |
| ) |
| |
| result = self.driver.ex_change_node_size(node=node, new_size=size) |
| self.assertTrue(result) |
| |
| def test_list_volumes(self): |
| volumes = self.driver.list_volumes() |
| |
| self.assertEqual(len(volumes), 3) |
| |
| self.assertEqual("vol-10ae5e2b", volumes[0].id) |
| self.assertEqual(1, volumes[0].size) |
| self.assertEqual("available", volumes[0].extra["state"]) |
| self.assertEqual(StorageVolumeState.AVAILABLE, volumes[0].state) |
| |
| self.assertEqual("vol-v24bfh75", volumes[1].id) |
| self.assertEqual(11, volumes[1].size) |
| self.assertIsNone(volumes[1].extra["snapshot_id"]) |
| self.assertEqual("in-use", volumes[1].extra["state"]) |
| self.assertEqual(StorageVolumeState.INUSE, volumes[1].state) |
| |
| self.assertEqual("vol-b6c851ec", volumes[2].id) |
| self.assertEqual(8, volumes[2].size) |
| self.assertEqual("some-unknown-status", volumes[2].extra["state"]) |
| self.assertEqual("i-d334b4b3", volumes[2].extra["instance_id"]) |
| self.assertEqual("/dev/sda1", volumes[2].extra["device"]) |
| self.assertEqual("snap-30d37269", volumes[2].extra["snapshot_id"]) |
| self.assertEqual(StorageVolumeState.UNKNOWN, volumes[2].state) |
| |
| EC2MockHttp.type = "filters_nodes" |
| node = Node("i-d334b4b3", None, None, None, None, self.driver) |
| self.driver.list_volumes(node=node) |
| |
| EC2MockHttp.type = "filters_status" |
| self.driver.list_volumes(ex_filters={"status": "available"}) |
| |
| def test_create_volume(self): |
| location = self.driver.list_locations()[0] |
| vol = self.driver.create_volume(10, "vol", location=location) |
| |
| self.assertEqual(10, vol.size) |
| self.assertEqual("vol", vol.name) |
| self.assertEqual("creating", vol.extra["state"]) |
| self.assertTrue(isinstance(vol.extra["create_time"], datetime)) |
| self.assertEqual(False, vol.extra["encrypted"]) |
| |
| expected_msg = "Invalid volume type specified: invalid-type" |
| self.assertRaisesRegex( |
| ValueError, |
| expected_msg, |
| self.driver.create_volume, |
| 10, |
| "invalid-vol", |
| location=location, |
| ex_volume_type="invalid-type", |
| ) |
| |
| def test_create_encrypted_volume(self): |
| location = self.driver.list_locations()[0] |
| vol = self.driver.create_volume( |
| 10, "vol", location, ex_encrypted=True, ex_kms_key_id="1234" |
| ) |
| |
| self.assertEqual(10, vol.size) |
| self.assertEqual("vol", vol.name) |
| self.assertEqual("creating", vol.extra["state"]) |
| self.assertTrue(isinstance(vol.extra["create_time"], datetime)) |
| self.assertEqual(True, vol.extra["encrypted"]) |
| |
| def test_destroy_volume(self): |
| vol = StorageVolume( |
| id="vol-4282672b", |
| name="test", |
| state=StorageVolumeState.AVAILABLE, |
| size=10, |
| driver=self.driver, |
| ) |
| |
| retValue = self.driver.destroy_volume(vol) |
| self.assertTrue(retValue) |
| |
| def test_attach(self): |
| vol = StorageVolume( |
| id="vol-4282672b", |
| name="test", |
| size=10, |
| state=StorageVolumeState.AVAILABLE, |
| driver=self.driver, |
| ) |
| |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| retValue = self.driver.attach_volume(node, vol, "/dev/sdh") |
| |
| self.assertTrue(retValue) |
| |
| def test_detach(self): |
| vol = StorageVolume( |
| id="vol-4282672b", |
| name="test", |
| state=StorageVolumeState.INUSE, |
| size=10, |
| driver=self.driver, |
| ) |
| |
| retValue = self.driver.detach_volume(vol) |
| self.assertTrue(retValue) |
| |
| def test_create_volume_snapshot(self): |
| vol = StorageVolume( |
| id="vol-4282672b", |
| name="test", |
| state=StorageVolumeState.AVAILABLE, |
| size=10, |
| driver=self.driver, |
| ) |
| snap = self.driver.create_volume_snapshot(vol, "Test snapshot") |
| self.assertEqual("snap-a7cb2hd9", snap.id) |
| self.assertEqual(vol.size, snap.size) |
| self.assertEqual("Test snapshot", snap.extra["name"]) |
| self.assertEqual(vol.id, snap.extra["volume_id"]) |
| self.assertEqual("pending", snap.extra["state"]) |
| self.assertEqual(VolumeSnapshotState.CREATING, snap.state) |
| # 2013-08-15T16:22:30.000Z |
| self.assertEqual(datetime(2013, 8, 15, 16, 22, 30, tzinfo=UTC), snap.created) |
| |
| def test_create_volume_snapshot_with_tags(self): |
| vol = StorageVolume( |
| id="vol-4282672b", |
| name="test", |
| state=StorageVolumeState.AVAILABLE, |
| size=10, |
| driver=self.driver, |
| ) |
| snap = self.driver.create_volume_snapshot( |
| vol, "Test snapshot", ex_metadata={"my_tag": "test"} |
| ) |
| self.assertEqual("test", snap.extra["tags"]["my_tag"]) |
| |
| def test_list_snapshots(self): |
| snaps = self.driver.list_snapshots() |
| |
| self.assertEqual(len(snaps), 3) |
| |
| self.assertEqual("snap-428abd35", snaps[0].id) |
| self.assertEqual(VolumeSnapshotState.CREATING, snaps[0].state) |
| self.assertEqual("vol-e020df80", snaps[0].extra["volume_id"]) |
| self.assertEqual(30, snaps[0].size) |
| self.assertEqual("Daily Backup", snaps[0].extra["description"]) |
| |
| self.assertEqual("snap-18349159", snaps[1].id) |
| self.assertEqual("DB Backup 1", snaps[1].name) |
| self.assertEqual(VolumeSnapshotState.AVAILABLE, snaps[1].state) |
| self.assertEqual("vol-b5a2c1v9", snaps[1].extra["volume_id"]) |
| self.assertEqual(15, snaps[1].size) |
| self.assertEqual("Weekly backup", snaps[1].extra["description"]) |
| self.assertEqual("DB Backup 1", snaps[1].extra["name"]) |
| |
| def test_list_volume_snapshots(self): |
| volume = self.driver.list_volumes()[0] |
| assert volume.id == "vol-10ae5e2b" |
| |
| snapshots = self.driver.list_volume_snapshots(volume) |
| self.assertEqual(len(snapshots), 1) |
| self.assertEqual(snapshots[0].id, "snap-18349160") |
| |
| def test_destroy_snapshot(self): |
| snap = VolumeSnapshot(id="snap-428abd35", size=10, driver=self.driver) |
| resp = snap.destroy() |
| self.assertTrue(resp) |
| |
| def test_ex_modify_image_attribute(self): |
| images = self.driver.list_images() |
| image = images[0] |
| |
| data = {"LaunchPermission.Add.1.Group": "all"} |
| resp = self.driver.ex_modify_image_attribute(image, data) |
| self.assertTrue(resp) |
| |
| def test_ex_modify_snapshot_attribute(self): |
| snap = VolumeSnapshot(id="snap-1234567890abcdef0", size=10, driver=self.driver) |
| |
| data = {"CreateVolumePermission.Add.1.Group": "all"} |
| resp = self.driver.ex_modify_snapshot_attribute(snap, data) |
| self.assertTrue(resp) |
| |
| def test_create_node_ex_security_groups(self): |
| EC2MockHttp.type = "ex_security_groups" |
| |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| security_groups = ["group1", "group2"] |
| |
| # Old, deprecated argument name |
| self.driver.create_node( |
| name="foo", image=image, size=size, ex_securitygroup=security_groups |
| ) |
| |
| # New argument name |
| self.driver.create_node( |
| name="foo", image=image, size=size, ex_security_groups=security_groups |
| ) |
| |
| # Test old and new arguments are mutually exclusive |
| self.assertRaises( |
| ValueError, |
| self.driver.create_node, |
| name="foo", |
| image=image, |
| size=size, |
| ex_securitygroup=security_groups, |
| ex_security_groups=security_groups, |
| ) |
| |
| def test_create_node_ex_security_group_ids(self): |
| EC2MockHttp.type = "ex_security_group_ids" |
| |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| subnet = EC2NetworkSubnet(12345, "test_subnet", "pending") |
| security_groups = ["sg-1aa11a1a", "sg-2bb22b2b"] |
| |
| self.driver.create_node( |
| name="foo", |
| image=image, |
| size=size, |
| ex_security_group_ids=security_groups, |
| ex_subnet=subnet, |
| ) |
| self.assertRaises( |
| ValueError, |
| self.driver.create_node, |
| name="foo", |
| image=image, |
| size=size, |
| ex_security_group_ids=security_groups, |
| ) |
| |
| def test_ex_get_metadata_for_node(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| node = self.driver.create_node( |
| name="foo", image=image, size=size, ex_metadata={"Bar": "baz", "Num": "42"} |
| ) |
| |
| metadata = self.driver.ex_get_metadata_for_node(node) |
| self.assertEqual(metadata["Name"], "foo") |
| self.assertEqual(metadata["Bar"], "baz") |
| self.assertEqual(metadata["Num"], "42") |
| self.assertEqual(len(metadata), 3) |
| |
| def test_ex_get_limits(self): |
| limits = self.driver.ex_get_limits() |
| |
| expected = {"max-instances": 20, "vpc-max-elastic-ips": 5, "max-elastic-ips": 5} |
| self.assertEqual(limits["resource"], expected) |
| |
| def test_ex_create_security_group(self): |
| group = self.driver.ex_create_security_group( |
| "WebServers", "Rules to protect web nodes", "vpc-143cab4" |
| ) |
| |
| self.assertEqual(group["group_id"], "sg-52e2f530") |
| |
| def test_ex_create_placement_groups(self): |
| resp = self.driver.ex_create_placement_group("NewPG") |
| self.assertTrue(resp) |
| |
| def test_ex_delete_placement_groups(self): |
| pgs = self.driver.ex_list_placement_groups() |
| pg = pgs[0] |
| |
| resp = self.driver.ex_delete_placement_group(pg.name) |
| self.assertTrue(resp) |
| |
| def test_ex_list_placement_groups(self): |
| pgs = self.driver.ex_list_placement_groups() |
| self.assertEqual(len(pgs), 2) |
| self.assertIsInstance(pgs[0], EC2PlacementGroup) |
| |
| def test_ex_list_networks(self): |
| vpcs = self.driver.ex_list_networks() |
| |
| self.assertEqual(len(vpcs), 2) |
| |
| self.assertEqual("vpc-532335e1", vpcs[0].id) |
| self.assertEqual("vpc-532335e1", vpcs[0].name) |
| self.assertEqual("192.168.51.0/24", vpcs[0].cidr_block) |
| self.assertEqual("available", vpcs[0].extra["state"]) |
| self.assertEqual("dopt-7eded312", vpcs[0].extra["dhcp_options_id"]) |
| |
| self.assertEqual("vpc-62ded30e", vpcs[1].id) |
| self.assertEqual("Test VPC", vpcs[1].name) |
| self.assertEqual("192.168.52.0/24", vpcs[1].cidr_block) |
| self.assertEqual("available", vpcs[1].extra["state"]) |
| self.assertEqual("dopt-7eded312", vpcs[1].extra["dhcp_options_id"]) |
| |
| def test_ex_list_networks_network_ids(self): |
| EC2MockHttp.type = "network_ids" |
| network_ids = ["vpc-532335e1"] |
| |
| # We assert in the mock http method |
| self.driver.ex_list_networks(network_ids=network_ids) |
| |
| def test_ex_list_networks_filters(self): |
| EC2MockHttp.type = "filters" |
| |
| filters = OrderedDict( |
| [ |
| ("dhcp-options-id", "dopt-7eded312"), # matches two networks |
| ("cidr", "192.168.51.0/24"), # matches two networks |
| ] |
| ) |
| |
| # We assert in the mock http method |
| self.driver.ex_list_networks(filters=filters) |
| |
| def test_ex_create_network(self): |
| vpc = self.driver.ex_create_network( |
| "192.168.55.0/24", name="Test VPC", instance_tenancy="default" |
| ) |
| |
| self.assertEqual("vpc-ad3527cf", vpc.id) |
| self.assertEqual("192.168.55.0/24", vpc.cidr_block) |
| self.assertEqual("pending", vpc.extra["state"]) |
| |
| def test_ex_delete_network(self): |
| vpcs = self.driver.ex_list_networks() |
| vpc = vpcs[0] |
| |
| resp = self.driver.ex_delete_network(vpc) |
| self.assertTrue(resp) |
| |
| def test_ex_list_subnets(self): |
| subnets = self.driver.ex_list_subnets() |
| |
| self.assertEqual(len(subnets), 2) |
| |
| self.assertEqual("subnet-ce0e7ce5", subnets[0].id) |
| self.assertEqual("available", subnets[0].state) |
| self.assertEqual(123, subnets[0].extra["available_ips"]) |
| |
| self.assertEqual("subnet-ce0e7ce6", subnets[1].id) |
| self.assertEqual("available", subnets[1].state) |
| self.assertEqual(59, subnets[1].extra["available_ips"]) |
| |
| def test_ex_create_subnet(self): |
| subnet = self.driver.ex_create_subnet( |
| "vpc-532135d1", "192.168.51.128/26", "us-east-1b", name="Test Subnet" |
| ) |
| |
| self.assertEqual("subnet-ce0e7ce6", subnet.id) |
| self.assertEqual("pending", subnet.state) |
| self.assertEqual("vpc-532135d1", subnet.extra["vpc_id"]) |
| |
| def test_ex_modify_subnet_attribute(self): |
| subnet = self.driver.ex_list_subnets()[0] |
| resp = self.driver.ex_modify_subnet_attribute(subnet, "auto_public_ip", True) |
| self.assertTrue(resp) |
| resp = self.driver.ex_modify_subnet_attribute(subnet, "auto_ipv6", False) |
| self.assertTrue(resp) |
| |
| expected_msg = "Unsupported attribute: invalid" |
| self.assertRaisesRegex( |
| ValueError, |
| expected_msg, |
| self.driver.ex_modify_subnet_attribute, |
| subnet, |
| "invalid", |
| True, |
| ) |
| |
| def test_ex_delete_subnet(self): |
| subnet = self.driver.ex_list_subnets()[0] |
| resp = self.driver.ex_delete_subnet(subnet=subnet) |
| self.assertTrue(resp) |
| |
| def test_ex_get_console_output(self): |
| node = self.driver.list_nodes()[0] |
| resp = self.driver.ex_get_console_output(node) |
| self.assertEqual("Test String", resp["output"]) |
| |
| def test_ex_list_network_interfaces(self): |
| interfaces = self.driver.ex_list_network_interfaces() |
| |
| self.assertEqual(len(interfaces), 2) |
| |
| self.assertEqual("eni-18e6c05e", interfaces[0].id) |
| self.assertEqual("in-use", interfaces[0].state) |
| self.assertEqual("0e:6e:df:72:78:af", interfaces[0].extra["mac_address"]) |
| |
| self.assertEqual("eni-83e3c5c5", interfaces[1].id) |
| self.assertEqual("in-use", interfaces[1].state) |
| self.assertEqual("0e:93:0b:e9:e9:c4", interfaces[1].extra["mac_address"]) |
| |
| def test_ex_create_network_interface(self): |
| subnet = self.driver.ex_list_subnets()[0] |
| interface = self.driver.ex_create_network_interface( |
| subnet, name="Test Interface", description="My Test" |
| ) |
| |
| self.assertEqual("eni-2b36086d", interface.id) |
| self.assertEqual("pending", interface.state) |
| self.assertEqual("0e:bd:49:3e:11:74", interface.extra["mac_address"]) |
| |
| def test_ex_delete_network_interface(self): |
| interface = self.driver.ex_list_network_interfaces()[0] |
| resp = self.driver.ex_delete_network_interface(interface) |
| self.assertTrue(resp) |
| |
| def test_ex_attach_network_interface_to_node(self): |
| node = self.driver.list_nodes()[0] |
| interface = self.driver.ex_list_network_interfaces()[0] |
| resp = self.driver.ex_attach_network_interface_to_node(interface, node, 1) |
| self.assertTrue(resp) |
| |
| def test_ex_detach_network_interface(self): |
| resp = self.driver.ex_detach_network_interface("eni-attach-2b588b47") |
| self.assertTrue(resp) |
| |
| def test_ex_list_internet_gateways(self): |
| gateways = self.driver.ex_list_internet_gateways() |
| |
| self.assertEqual(len(gateways), 2) |
| |
| self.assertEqual("igw-84dd3ae1", gateways[0].id) |
| self.assertEqual("igw-7fdae215", gateways[1].id) |
| self.assertEqual("available", gateways[1].state) |
| self.assertEqual("vpc-62cad41e", gateways[1].vpc_id) |
| |
| def test_ex_create_internet_gateway(self): |
| gateway = self.driver.ex_create_internet_gateway() |
| |
| self.assertEqual("igw-13ac2b36", gateway.id) |
| |
| def test_ex_delete_internet_gateway(self): |
| gateway = self.driver.ex_list_internet_gateways()[0] |
| resp = self.driver.ex_delete_internet_gateway(gateway) |
| self.assertTrue(resp) |
| |
| def test_ex_attach_internet_gateway(self): |
| gateway = self.driver.ex_list_internet_gateways()[0] |
| network = self.driver.ex_list_networks()[0] |
| resp = self.driver.ex_attach_internet_gateway(gateway, network) |
| self.assertTrue(resp) |
| |
| def test_ex_detach_internet_gateway(self): |
| gateway = self.driver.ex_list_internet_gateways()[0] |
| network = self.driver.ex_list_networks()[0] |
| resp = self.driver.ex_detach_internet_gateway(gateway, network) |
| self.assertTrue(resp) |
| |
| def test_ex_modify_volume(self): |
| volume = self.driver.list_volumes()[0] |
| assert volume.id == "vol-10ae5e2b" |
| |
| params = {"VolumeType": "io1", "Size": 2, "Iops": 1000} |
| volume_modification = self.driver.ex_modify_volume(volume, params) |
| |
| self.assertIsNone(volume_modification.end_time) |
| self.assertEqual("modifying", volume_modification.modification_state) |
| self.assertEqual(300, volume_modification.original_iops) |
| self.assertEqual(1, volume_modification.original_size) |
| self.assertEqual("gp2", volume_modification.original_volume_type) |
| self.assertEqual(0, volume_modification.progress) |
| self.assertIsNone(volume_modification.status_message) |
| self.assertEqual(1000, volume_modification.target_iops) |
| self.assertEqual(2, volume_modification.target_size) |
| self.assertEqual("io1", volume_modification.target_volume_type) |
| self.assertEqual("vol-10ae5e2b", volume_modification.volume_id) |
| |
| def test_ex_describe_volumes_modifications(self): |
| modifications = self.driver.ex_describe_volumes_modifications() |
| |
| self.assertEqual(len(modifications), 2) |
| |
| self.assertIsNone(modifications[0].end_time) |
| self.assertEqual("optimizing", modifications[0].modification_state) |
| self.assertEqual(100, modifications[0].original_iops) |
| self.assertEqual(10, modifications[0].original_size) |
| self.assertEqual("gp2", modifications[0].original_volume_type) |
| self.assertEqual(3, modifications[0].progress) |
| self.assertIsNone(modifications[0].status_message) |
| self.assertEqual(10000, modifications[0].target_iops) |
| self.assertEqual(2000, modifications[0].target_size) |
| self.assertEqual("io1", modifications[0].target_volume_type) |
| self.assertEqual("vol-06397e7a0eEXAMPLE", modifications[0].volume_id) |
| |
| self.assertEqual("completed", modifications[1].modification_state) |
| self.assertEqual(100, modifications[1].original_iops) |
| self.assertEqual(8, modifications[1].original_size) |
| self.assertEqual("gp2", modifications[1].original_volume_type) |
| self.assertEqual(100, modifications[1].progress) |
| self.assertIsNone(modifications[1].status_message) |
| self.assertEqual(10000, modifications[1].target_iops) |
| self.assertEqual(200, modifications[1].target_size) |
| self.assertEqual("io1", modifications[1].target_volume_type) |
| self.assertEqual("vol-bEXAMPLE", modifications[1].volume_id) |
| |
| def test_params_is_not_simple_type_exception_is_thrown(self): |
| params = {"not": {"not": ["simple"]}} |
| |
| expected_msg = 'dictionary contains an attribute "not" which value' |
| self.assertRaisesRegex( |
| ValueError, expected_msg, self.driver.connection.request, "/", params=params |
| ) |
| |
| params = {"invalid": [1, 2, 3]} |
| |
| expected_msg = 'dictionary contains an attribute "invalid" which value' |
| self.assertRaisesRegex( |
| ValueError, expected_msg, self.driver.connection.request, "/", params=params |
| ) |
| |
| |
| class EC2USWest1Tests(EC2Tests): |
| region = "us-west-1" |
| |
| |
| class EC2USWest2Tests(EC2Tests): |
| region = "us-west-2" |
| |
| |
| class EC2EUWestTests(EC2Tests): |
| region = "eu-west-1" |
| |
| |
| class EC2APSE1Tests(EC2Tests): |
| region = "ap-southeast-1" |
| |
| |
| class EC2APNETests(EC2Tests): |
| region = "ap-northeast-1" |
| |
| |
| class EC2APSE2Tests(EC2Tests): |
| region = "ap-southeast-2" |
| |
| |
| class EC2SAEastTests(EC2Tests): |
| region = "sa-east-1" |
| |
| |
| class EC2MockHttp(MockHttp, unittest.TestCase): |
| fixtures = ComputeFileFixtures("ec2") |
| |
| def _DescribeInstances(self, method, url, body, headers): |
| body = self.fixtures.load("describe_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeReservedInstances(self, method, url, body, headers): |
| body = self.fixtures.load("describe_reserved_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeAvailabilityZones(self, method, url, body, headers): |
| body = self.fixtures.load("describe_availability_zones.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _RebootInstances(self, method, url, body, headers): |
| body = self.fixtures.load("reboot_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _StartInstances(self, method, url, body, headers): |
| body = self.fixtures.load("start_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _StopInstances(self, method, url, body, headers): |
| body = self.fixtures.load("stop_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeSecurityGroups(self, method, url, body, headers): |
| body = self.fixtures.load("describe_security_groups.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteSecurityGroup(self, method, url, body, headers): |
| body = self.fixtures.load("delete_security_group.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AuthorizeSecurityGroupIngress(self, method, url, body, headers): |
| body = self.fixtures.load("authorize_security_group_ingress.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeImages(self, method, url, body, headers): |
| body = self.fixtures.load("describe_images.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _RegisterImages(self, method, url, body, headers): |
| body = self.fixtures.load("register_image.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ImportSnapshot(self, method, url, body, headers): |
| body = self.fixtures.load("import_snapshot.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeImportSnapshotTasks(self, method, url, body, headers): |
| body = self.fixtures.load("describe_import_snapshot_tasks.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _timeout_DescribeImportSnapshotTasks(self, method, url, body, headers): |
| body = self.fixtures.load("describe_import_snapshot_tasks_active.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ex_imageids_DescribeImages(self, method, url, body, headers): |
| body = self.fixtures.load("describe_images_ex_imageids.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _RunInstances(self, method, url, body, headers): |
| body = self.fixtures.load("run_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ex_user_data_RunInstances(self, method, url, body, headers): |
| # test_create_node_with_ex_userdata |
| if url.startswith("/"): |
| url = url[1:] |
| |
| if url.startswith("?"): |
| url = url[1:] |
| |
| params = parse_qs(url) |
| |
| self.assertTrue( |
| "UserData" in params, |
| "UserData not in params, actual params: %s" % (str(params)), |
| ) |
| user_data = base64.b64decode(b(params["UserData"][0])).decode("utf-8") |
| self.assertEqual(user_data, "foo\nbar\foo") |
| |
| body = self.fixtures.load("run_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_ex_assign_public_ip_RunInstances(self, method, url, body, headers): |
| self.assertUrlContainsQueryParams( |
| url, |
| { |
| "NetworkInterface.1.AssociatePublicIpAddress": "true", |
| "NetworkInterface.1.DeleteOnTermination": "true", |
| "NetworkInterface.1.DeviceIndex": "0", |
| "NetworkInterface.1.SubnetId": "subnet-11111111", |
| "NetworkInterface.1.SecurityGroupId.1": "sg-11111111", |
| }, |
| ) |
| body = self.fixtures.load("run_instances_with_subnet_and_security_group.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_ex_terminate_on_shutdown_RunInstances(self, method, url, body, headers): |
| self.assertUrlContainsQueryParams(url, {"InstanceInitiatedShutdownBehavior": "terminate"}) |
| |
| body = self.fixtures.load("run_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ex_security_groups_RunInstances(self, method, url, body, headers): |
| self.assertUrlContainsQueryParams(url, {"SecurityGroup.1": "group1"}) |
| self.assertUrlContainsQueryParams(url, {"SecurityGroup.2": "group2"}) |
| |
| body = self.fixtures.load("run_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ex_security_group_ids_RunInstances(self, method, url, body, headers): |
| self.assertUrlContainsQueryParams(url, {"SecurityGroupId.1": "sg-1aa11a1a"}) |
| self.assertUrlContainsQueryParams(url, {"SecurityGroupId.2": "sg-2bb22b2b"}) |
| |
| body = self.fixtures.load("run_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_ex_blockdevicemappings_RunInstances(self, method, url, body, headers): |
| expected_params = { |
| "BlockDeviceMapping.1.DeviceName": "/dev/sda1", |
| "BlockDeviceMapping.1.Ebs.VolumeSize": "10", |
| "BlockDeviceMapping.2.DeviceName": "/dev/sdb", |
| "BlockDeviceMapping.2.VirtualName": "ephemeral0", |
| "BlockDeviceMapping.3.DeviceName": "/dev/sdc", |
| "BlockDeviceMapping.3.VirtualName": "ephemeral1", |
| } |
| self.assertUrlContainsQueryParams(url, expected_params) |
| |
| body = self.fixtures.load("run_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _idempotent_RunInstances(self, method, url, body, headers): |
| body = self.fixtures.load("run_instances_idem.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _idempotent_mismatch_RunInstances(self, method, url, body, headers): |
| body = self.fixtures.load("run_instances_idem_mismatch.xml") |
| return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.BAD_REQUEST]) |
| |
| def _ex_iam_profile_RunInstances(self, method, url, body, headers): |
| body = self.fixtures.load("run_instances_iam_profile.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ex_spot_RunInstances(self, method, url, body, headers): |
| body = self.fixtures.load("run_instances_spot.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _TerminateInstances(self, method, url, body, headers): |
| body = self.fixtures.load("terminate_instances.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeKeyPairs(self, method, url, body, headers): |
| body = self.fixtures.load("describe_key_pairs.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _get_one_DescribeKeyPairs(self, method, url, body, headers): |
| self.assertUrlContainsQueryParams(url, {"KeyName": "gsg-keypair"}) |
| |
| body = self.fixtures.load("describe_key_pairs.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _doesnt_exist_DescribeKeyPairs(self, method, url, body, headers): |
| body = self.fixtures.load("describe_key_pairs_doesnt_exist.xml") |
| return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.BAD_REQUEST]) |
| |
| def _CreateKeyPair(self, method, url, body, headers): |
| body = self.fixtures.load("create_key_pair.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ImportKeyPair(self, method, url, body, headers): |
| body = self.fixtures.load("import_key_pair.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeTags(self, method, url, body, headers): |
| body = self.fixtures.load("describe_tags.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateTags(self, method, url, body, headers): |
| body = self.fixtures.load("create_tags.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteTags(self, method, url, body, headers): |
| body = self.fixtures.load("delete_tags.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeAddresses(self, method, url, body, headers): |
| body = self.fixtures.load("describe_addresses_multi.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AllocateAddress(self, method, url, body, headers): |
| body = self.fixtures.load("allocate_address.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _vpc_AllocateAddress(self, method, url, body, headers): |
| body = self.fixtures.load("allocate_vpc_address.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AssociateAddress(self, method, url, body, headers): |
| body = self.fixtures.load("associate_address.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _vpc_AssociateAddress(self, method, url, body, headers): |
| body = self.fixtures.load("associate_vpc_address.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DisassociateAddress(self, method, url, body, headers): |
| body = self.fixtures.load("disassociate_address.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ReleaseAddress(self, method, url, body, headers): |
| body = self.fixtures.load("release_address.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _all_addresses_DescribeAddresses(self, method, url, body, headers): |
| body = self.fixtures.load("describe_addresses_all.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _WITH_TAGS_DescribeAddresses(self, method, url, body, headers): |
| body = self.fixtures.load("describe_addresses_multi.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifyInstanceAttribute(self, method, url, body, headers): |
| body = self.fixtures.load("modify_instance_attribute.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifySnapshotAttribute(self, method, url, body, headers): |
| body = self.fixtures.load("modify_snapshot_attribute.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateVolume(self, method, url, body, headers): |
| if "KmsKeyId=" in url: |
| assert "Encrypted=1" in url, ( |
| "If a KmsKeyId is specified, the " "Encrypted flag must also be set." |
| ) |
| if "Encrypted=1" in url: |
| body = self.fixtures.load("create_encrypted_volume.xml") |
| else: |
| body = self.fixtures.load("create_volume.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteVolume(self, method, url, body, headers): |
| body = self.fixtures.load("delete_volume.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AttachVolume(self, method, url, body, headers): |
| body = self.fixtures.load("attach_volume.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DetachVolume(self, method, url, body, headers): |
| body = self.fixtures.load("detach_volume.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeVolumes(self, method, url, body, headers): |
| body = self.fixtures.load("describe_volumes.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _filters_nodes_DescribeVolumes(self, method, url, body, headers): |
| expected_params = { |
| "Filter.1.Name": "attachment.instance-id", |
| "Filter.1.Value.1": "i-d334b4b3", |
| } |
| |
| self.assertUrlContainsQueryParams(url, expected_params) |
| |
| body = self.fixtures.load("describe_volumes.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _filters_status_DescribeVolumes(self, method, url, body, headers): |
| expected_params = {"Filter.1.Name": "status", "Filter.1.Value.1": "available"} |
| |
| self.assertUrlContainsQueryParams(url, expected_params) |
| |
| body = self.fixtures.load("describe_volumes.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateSnapshot(self, method, url, body, headers): |
| body = self.fixtures.load("create_snapshot.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeSnapshots(self, method, url, body, headers): |
| body = self.fixtures.load("describe_snapshots.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteSnapshot(self, method, url, body, headers): |
| body = self.fixtures.load("delete_snapshot.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CopyImage(self, method, url, body, headers): |
| body = self.fixtures.load("copy_image.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateImage(self, method, url, body, headers): |
| body = self.fixtures.load("create_image.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeregisterImage(self, method, url, body, headers): |
| body = self.fixtures.load("deregister_image.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteKeyPair(self, method, url, body, headers): |
| self.assertUrlContainsQueryParams(url, {"KeyName": "gsg-keypair"}) |
| |
| body = self.fixtures.load("delete_key_pair.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifyImageAttribute(self, method, url, body, headers): |
| body = self.fixtures.load("modify_image_attribute.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeAccountAttributes(self, method, url, body, headers): |
| body = self.fixtures.load("describe_account_attributes.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateSecurityGroup(self, method, url, body, headers): |
| body = self.fixtures.load("create_security_group.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeVpcs(self, method, url, body, headers): |
| body = self.fixtures.load("describe_vpcs.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _network_ids_DescribeVpcs(self, method, url, body, headers): |
| expected_params = {"VpcId.1": "vpc-532335e1"} |
| self.assertUrlContainsQueryParams(url, expected_params) |
| |
| body = self.fixtures.load("describe_vpcs.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _filters_DescribeVpcs(self, method, url, body, headers): |
| expected_params = { |
| "Filter.1.Name": "dhcp-options-id", |
| "Filter.1.Value.1": "dopt-7eded312", |
| "Filter.2.Name": "cidr", |
| "Filter.2.Value.1": "192.168.51.0/24", |
| } |
| |
| self.assertUrlContainsQueryParams(url, expected_params) |
| |
| body = self.fixtures.load("describe_vpcs.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateVpc(self, method, url, body, headers): |
| body = self.fixtures.load("create_vpc.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteVpc(self, method, url, body, headers): |
| body = self.fixtures.load("delete_vpc.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeSubnets(self, method, url, body, headers): |
| body = self.fixtures.load("describe_subnets.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateSubnet(self, method, url, body, headers): |
| body = self.fixtures.load("create_subnet.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifySubnetAttribute(self, method, url, body, headers): |
| body = self.fixtures.load("modify_subnet_attribute.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteSubnet(self, method, url, body, headers): |
| body = self.fixtures.load("delete_subnet.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _GetConsoleOutput(self, method, url, body, headers): |
| body = self.fixtures.load("get_console_output.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeNetworkInterfaces(self, method, url, body, headers): |
| body = self.fixtures.load("describe_network_interfaces.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateNetworkInterface(self, method, url, body, headers): |
| body = self.fixtures.load("create_network_interface.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteNetworkInterface(self, method, url, body, headers): |
| body = self.fixtures.load("delete_network_interface.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AttachNetworkInterface(self, method, url, body, headers): |
| body = self.fixtures.load("attach_network_interface.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DetachNetworkInterface(self, method, url, body, headers): |
| body = self.fixtures.load("detach_network_interface.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeInternetGateways(self, method, url, body, headers): |
| body = self.fixtures.load("describe_internet_gateways.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateInternetGateway(self, method, url, body, headers): |
| body = self.fixtures.load("create_internet_gateway.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteInternetGateway(self, method, url, body, headers): |
| body = self.fixtures.load("delete_internet_gateway.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AttachInternetGateway(self, method, url, body, headers): |
| body = self.fixtures.load("attach_internet_gateway.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DetachInternetGateway(self, method, url, body, headers): |
| body = self.fixtures.load("detach_internet_gateway.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreatePlacementGroup(self, method, url, body, headers): |
| body = self.fixtures.load("create_placement_groups.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeletePlacementGroup(self, method, url, body, headers): |
| body = self.fixtures.load("delete_placement_groups.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribePlacementGroups(self, method, url, body, headers): |
| body = self.fixtures.load("describe_placement_groups.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifyVolume(self, method, url, body, headers): |
| body = self.fixtures.load("modify_volume.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeVolumesModifications(self, method, url, body, headers): |
| body = self.fixtures.load("describe_volumes_modifications.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| |
| class EucMockHttp(EC2MockHttp): |
| fixtures = ComputeFileFixtures("ec2") |
| |
| def _services_Eucalyptus_DescribeInstances(self, method, url, body, headers): |
| return self._DescribeInstances(method, url, body, headers) |
| |
| def _services_Eucalyptus_DescribeImages(self, method, url, body, headers): |
| return self._DescribeImages(method, url, body, headers) |
| |
| def _services_Eucalyptus_DescribeAddresses(self, method, url, body, headers): |
| return self._DescribeAddresses(method, url, body, headers) |
| |
| def _services_Eucalyptus_RebootInstances(self, method, url, body, headers): |
| return self._RebootInstances(method, url, body, headers) |
| |
| def _services_Eucalyptus_TerminateInstances(self, method, url, body, headers): |
| return self._TerminateInstances(method, url, body, headers) |
| |
| def _services_Eucalyptus_RunInstances(self, method, url, body, headers): |
| return self._RunInstances(method, url, body, headers) |
| |
| def _services_Eucalyptus_DescribeInstanceTypes(self, method, url, body, headers): |
| body = self.fixtures.load("describe_instance_types.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| |
| class NimbusTests(EC2Tests): |
| def setUp(self): |
| NimbusNodeDriver.connectionCls.conn_class = EC2MockHttp |
| EC2MockHttp.use_param = "Action" |
| EC2MockHttp.type = None |
| self.driver = NimbusNodeDriver( |
| key=EC2_PARAMS[0], secret=EC2_PARAMS[1], host="some.nimbuscloud.com" |
| ) |
| |
| def test_ex_describe_addresses_for_node(self): |
| # overridden from EC2Tests -- Nimbus doesn't support elastic IPs. |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| ip_addresses = self.driver.ex_describe_addresses_for_node(node) |
| self.assertEqual(len(ip_addresses), 0) |
| |
| def test_ex_describe_addresses(self): |
| # overridden from EC2Tests -- Nimbus doesn't support elastic IPs. |
| node = Node("i-4382922a", None, None, None, None, self.driver) |
| nodes_elastic_ips = self.driver.ex_describe_addresses([node]) |
| |
| self.assertEqual(len(nodes_elastic_ips), 1) |
| self.assertEqual(len(nodes_elastic_ips[node.id]), 0) |
| |
| def test_list_sizes(self): |
| sizes = self.driver.list_sizes() |
| |
| ids = [s.id for s in sizes] |
| self.assertTrue("m1.small" in ids) |
| self.assertTrue("m1.large" in ids) |
| self.assertTrue("m1.xlarge" in ids) |
| |
| def test_list_nodes(self): |
| # overridden from EC2Tests -- Nimbus doesn't support elastic IPs. |
| node = self.driver.list_nodes()[0] |
| self.assertExecutedMethodCount(0) |
| public_ips = node.public_ips |
| self.assertEqual(node.id, "i-4382922a") |
| self.assertEqual(len(node.public_ips), 1) |
| self.assertEqual(public_ips[0], "1.2.3.4") |
| self.assertEqual(node.extra["tags"], {}) |
| |
| node = self.driver.list_nodes()[1] |
| self.assertExecutedMethodCount(0) |
| public_ips = node.public_ips |
| self.assertEqual(node.id, "i-8474834a") |
| self.assertEqual(len(node.public_ips), 1) |
| self.assertEqual(public_ips[0], "1.2.3.5") |
| self.assertEqual(node.extra["tags"], {"Name": "Test Server 2", "Group": "VPC Test"}) |
| |
| def test_ex_create_tags(self): |
| # Nimbus doesn't support creating tags so this one should be a |
| # passthrough |
| node = self.driver.list_nodes()[0] |
| self.driver.ex_create_tags(resource=node, tags={"foo": "bar"}) |
| self.assertExecutedMethodCount(0) |
| |
| def test_create_volume_snapshot_with_tags(self): |
| vol = StorageVolume( |
| id="vol-4282672b", |
| name="test", |
| state=StorageVolumeState.AVAILABLE, |
| size=10, |
| driver=self.driver, |
| ) |
| snap = self.driver.create_volume_snapshot( |
| vol, "Test snapshot", ex_metadata={"my_tag": "test"} |
| ) |
| self.assertDictEqual({}, snap.extra["tags"]) |
| |
| |
| class EucTests(LibcloudTestCase, TestCaseMixin): |
| def setUp(self): |
| EucNodeDriver.connectionCls.conn_class = EucMockHttp |
| EC2MockHttp.use_param = "Action" |
| EC2MockHttp.type = None |
| self.driver = EucNodeDriver( |
| key=EC2_PARAMS[0], |
| secret=EC2_PARAMS[1], |
| host="some.eucalyptus.com", |
| api_version="3.4.1", |
| ) |
| |
| def test_list_locations_response(self): |
| try: |
| self.driver.list_locations() |
| except Exception: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_list_location(self): |
| pass |
| |
| def test_list_sizes(self): |
| sizes = self.driver.list_sizes() |
| ids = [s.id for s in sizes] |
| self.assertEqual(len(ids), 18) |
| self.assertTrue("t1.micro" in ids) |
| self.assertTrue("m1.medium" in ids) |
| self.assertTrue("m3.xlarge" in ids) |
| |
| |
| class OutscaleTests(EC2Tests): |
| def setUp(self): |
| OutscaleSASNodeDriver.connectionCls.conn_class = EC2MockHttp |
| EC2MockHttp.use_param = "Action" |
| EC2MockHttp.type = None |
| self.driver = OutscaleSASNodeDriver( |
| key=EC2_PARAMS[0], secret=EC2_PARAMS[1], host="some.outscalecloud.com" |
| ) |
| |
| def test_ex_create_network(self): |
| # overridden from EC2Tests -- Outscale don't support instance_tenancy |
| vpc = self.driver.ex_create_network("192.168.55.0/24", name="Test VPC") |
| |
| self.assertEqual("vpc-ad3527cf", vpc.id) |
| self.assertEqual("192.168.55.0/24", vpc.cidr_block) |
| self.assertEqual("pending", vpc.extra["state"]) |
| |
| def test_ex_copy_image(self): |
| # overridden from EC2Tests -- Outscale does not support copying images |
| image = self.driver.list_images()[0] |
| try: |
| self.driver.ex_copy_image( |
| "us-east-1", image, name="Faux Image", description="Test Image Copy" |
| ) |
| except NotImplementedError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_ex_get_limits(self): |
| # overridden from EC2Tests -- Outscale does not support getting limits |
| try: |
| self.driver.ex_get_limits() |
| except NotImplementedError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_ex_create_network_interface(self): |
| # overridden from EC2Tests -- Outscale don't allow creating interfaces |
| subnet = self.driver.ex_list_subnets()[0] |
| try: |
| self.driver.ex_create_network_interface( |
| subnet, name="Test Interface", description="My Test" |
| ) |
| except NotImplementedError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_ex_delete_network_interface(self): |
| # overridden from EC2Tests -- Outscale don't allow deleting interfaces |
| interface = self.driver.ex_list_network_interfaces()[0] |
| try: |
| self.driver.ex_delete_network_interface(interface) |
| except NotImplementedError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_ex_attach_network_interface_to_node(self): |
| # overridden from EC2Tests -- Outscale don't allow attaching interfaces |
| node = self.driver.list_nodes()[0] |
| interface = self.driver.ex_list_network_interfaces()[0] |
| try: |
| self.driver.ex_attach_network_interface_to_node(interface, node, 1) |
| except NotImplementedError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_ex_detach_network_interface(self): |
| # overridden from EC2Tests -- Outscale don't allow detaching interfaces |
| try: |
| self.driver.ex_detach_network_interface("eni-attach-2b588b47") |
| except NotImplementedError: |
| pass |
| else: |
| self.fail("Exception was not thrown") |
| |
| def test_list_sizes(self): |
| sizes = self.driver.list_sizes() |
| |
| ids = [s.id for s in sizes] |
| self.assertTrue("m1.small" in ids) |
| self.assertTrue("m1.large" in ids) |
| self.assertTrue("m1.xlarge" in ids) |
| |
| def test_ex_create_node_with_ex_iam_profile(self): |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| self.assertRaises( |
| NotImplementedError, |
| self.driver.create_node, |
| name="foo", |
| image=image, |
| size=size, |
| ex_iamprofile="foo", |
| ) |
| |
| def test_create_node_with_ex_userdata(self): |
| EC2MockHttp.type = "ex_user_data" |
| |
| image = NodeImage(id="ami-be3adfd7", name=self.image_name, driver=self.driver) |
| size = NodeSize("m1.small", "Small Instance", None, None, None, None, driver=self.driver) |
| |
| result = self.driver.create_node( |
| name="foo", image=image, size=size, ex_userdata="foo\nbar\foo" |
| ) |
| self.assertTrue(result) |
| |
| |
| class FCUMockHttp(EC2MockHttp): |
| fixtures = ComputeFileFixtures("fcu") |
| |
| def _DescribeQuotas(self, method, url, body, headers): |
| body = self.fixtures.load("ex_describe_quotas.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeProductTypes(self, method, url, body, headers): |
| body = self.fixtures.load("ex_describe_product_types.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeInstanceTypes(self, method, url, body, headers): |
| body = self.fixtures.load("ex_describe_instance_types.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _GetProductType(self, method, url, body, headers): |
| body = self.fixtures.load("ex_get_product_type.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifyInstanceKeypair(self, method, url, body, headers): |
| body = self.fixtures.load("ex_modify_instance_keypair.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| |
| class OutscaleFCUTests(LibcloudTestCase): |
| def setUp(self): |
| OutscaleSASNodeDriver.connectionCls.conn_class = FCUMockHttp |
| EC2MockHttp.use_param = "Action" |
| EC2MockHttp.type = None |
| self.driver = OutscaleSASNodeDriver( |
| key=EC2_PARAMS[0], secret=EC2_PARAMS[1], host="some.fcucloud.com" |
| ) |
| |
| def test_ex_describe_quotas(self): |
| is_truncated, quota = self.driver.ex_describe_quotas() |
| self.assertTrue(is_truncated == "true") |
| self.assertTrue("global" in quota.keys()) |
| self.assertTrue("vpc-00000000" in quota.keys()) |
| |
| def test_ex_describe_product_types(self): |
| product_types = self.driver.ex_describe_product_types() |
| pt = {} |
| for e in product_types: |
| pt[e["productTypeId"]] = e["description"] |
| self.assertTrue("0001" in pt.keys()) |
| self.assertTrue("MapR" in pt.values()) |
| self.assertTrue(pt["0002"] == "Windows") |
| |
| def test_ex_describe_instance_instance_types(self): |
| instance_types = self.driver.ex_describe_instance_types() |
| it = {} |
| for e in instance_types: |
| it[e["name"]] = e["memory"] |
| self.assertTrue("og4.4xlarge" in it.keys()) |
| self.assertTrue("oc2.8xlarge" in it.keys()) |
| self.assertTrue("68718428160" in it.values()) |
| self.assertTrue(it["m3.large"] == "8050966528") |
| |
| def test_ex_get_product_type(self): |
| product_type = self.driver.ex_get_product_type("ami-29ab9e54") |
| self.assertTrue(product_type["productTypeId"] == "0002") |
| self.assertTrue(product_type["description"] == "Windows") |
| |
| def test_ex_modify_instance_keypair(self): |
| r = self.driver.ex_modify_instance_keypair("i-57292bc5", "key_name") |
| self.assertTrue(r) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |