| # 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 sys |
| import unittest |
| |
| from libcloud.test import MockHttp, LibcloudTestCase |
| from libcloud.utils.py3 import httplib |
| from libcloud.common.types import LibcloudError |
| from libcloud.compute.base import ( |
| Node, |
| NodeSize, |
| NodeImage, |
| NodeLocation, |
| StorageVolume, |
| VolumeSnapshot, |
| NodeAuthPassword, |
| ) |
| from libcloud.test.secrets import ECS_PARAMS |
| from libcloud.compute.types import NodeState, StorageVolumeState |
| from libcloud.test.file_fixtures import ComputeFileFixtures |
| from libcloud.compute.drivers.ecs import ECSDriver |
| |
| |
| class ECSDriverTestCase(LibcloudTestCase): |
| region = "cn-qingdao" |
| zone = "cn-qingdao-b" |
| image_id = "ubuntu1404_64_20G_aliaegis_20150325.vhd" |
| |
| def setUp(self): |
| ECSMockHttp.test = self |
| ECSDriver.connectionCls.conn_class = ECSMockHttp |
| ECSMockHttp.use_param = "Action" |
| ECSMockHttp.type = None |
| |
| self.driver = ECSDriver(*ECS_PARAMS, region=self.region) |
| self.fake_size = NodeSize( |
| "ecs.t1.small", "ecs t1 small", None, None, None, None, self.driver |
| ) |
| self.fake_image = NodeImage(self.image_id, name="ubuntu 14.04 64bit", driver=self.driver) |
| self.fake_node = Node( |
| id="fake-node1", |
| name="fake-node", |
| state=NodeState.RUNNING, |
| public_ips=None, |
| private_ips=None, |
| driver=self.driver, |
| ) |
| self.fake_volume = StorageVolume( |
| id="fake-volume1", |
| name="fake-volume", |
| size=self.fake_size, |
| driver=self.driver, |
| ) |
| self.fake_snapshot = VolumeSnapshot(id="fake-snapshot1", driver=self.driver) |
| self.fake_location = NodeLocation( |
| id=self.region, name=self.region, country=None, driver=self.driver |
| ) |
| self.fake_instance_id = "fake_instance_id" |
| self.fake_security_group_id = "fake_security_group_id" |
| |
| def test_list_nodes(self): |
| nodes = self.driver.list_nodes() |
| self.assertIsNotNone(nodes) |
| self.assertEqual(1, len(nodes)) |
| node = nodes[0] |
| self.assertEqual("iZ28n7dkvovZ", node.name) |
| self.assertEqual("i-28n7dkvov", node.id) |
| self.assertEqual(NodeState.PENDING, node.state) |
| self.assertEqual(1, len(node.public_ips)) |
| self.assertEqual("114.215.124.73", node.public_ips[0]) |
| self.assertEqual(1, len(node.private_ips)) |
| self.assertEqual("10.163.197.74", node.private_ips[0]) |
| expected_extra = { |
| "image_id": "ubuntu1404_64_20G_aliaegis_20150325.vhd", |
| "description": "", |
| "instance_type_family": "ecs.t1", |
| "zone_id": "cn-qingdao-b", |
| "internet_charge_type": "PayByTraffic", |
| "serial_number": "ca0122d9-374d-4fce-9fc0-71f7c3eaf1c3", |
| "io_optimized": "false", |
| "device_available": "true", |
| "instance_network_type": "classic", |
| "hostname": "iZ28n7dkvovZ", |
| "instance_type": "ecs.t1.small", |
| "creation_time": "2015-12-27T07:35Z", |
| "instance_charge_type": "PostPaid", |
| "expired_time": "2999-09-08T16:00Z", |
| } |
| self._validate_extras(expected_extra, node.extra) |
| vpc = { |
| "vpc_id": "", |
| "vswitch_id": "", |
| "private_ip_address": None, |
| "nat_ip_address": "", |
| } |
| self._validate_extras(vpc, node.extra["vpc_attributes"]) |
| eip_address = { |
| "allocation_id": "", |
| "ip_address": "", |
| "internet_charge_type": "", |
| "bandwidth": None, |
| } |
| self._validate_extras(eip_address, node.extra["eip_address"]) |
| self.assertIsNone(node.extra["operation_locks"]["lock_reason"]) |
| |
| def test_list_nodes_with_ex_node_ids(self): |
| ECSMockHttp.type = "list_nodes_ex_node_ids" |
| nodes = self.driver.list_nodes(ex_node_ids=["i-28n7dkvov", "not-existed-id"]) |
| self.assertIsNotNone(nodes) |
| |
| def test_list_nodes_with_ex_filters(self): |
| ECSMockHttp.type = "list_nodes_ex_filters" |
| nodes = self.driver.list_nodes(ex_filters={"ZoneId": self.zone}) |
| self.assertIsNotNone(nodes) |
| |
| def _validate_extras(self, expected, actual): |
| self.assertIsNotNone(actual) |
| for key, value in iter(expected.items()): |
| self.assertTrue(key in actual) |
| self.assertEqual( |
| value, |
| actual[key], |
| ( |
| "extra %(key)s not equal, " |
| 'expected: "%(expected)s", ' |
| 'actual: "%(actual)s"' % {"key": key, "expected": value, "actual": actual[key]} |
| ), |
| ) |
| |
| def test_create_node(self): |
| ECSMockHttp.type = "create_node" |
| name = "test_create_node" |
| node = self.driver.create_node( |
| name=name, |
| image=self.fake_image, |
| size=self.fake_size, |
| ex_security_group_id="sg-28ou0f3xa", |
| ex_description="description", |
| ex_internet_charge_type="PayByTraffic", |
| ex_internet_max_bandwidth_out=1, |
| ex_internet_max_bandwidth_in=200, |
| ex_hostname="hostname", |
| auth=NodeAuthPassword("password"), |
| ex_io_optimized=True, |
| ex_system_disk={ |
| "category": "cloud", |
| "disk_name": "root", |
| "description": "sys", |
| }, |
| ex_vswitch_id="vswitch-id1", |
| ex_private_ip_address="1.1.1.2", |
| ex_client_token="client_token", |
| ) |
| self.assertIsNotNone(node) |
| |
| def test_create_node_with_data_disk(self): |
| ECSMockHttp.type = "create_node_with_data" |
| self.name = "test_create_node" |
| self.data_disk = { |
| "size": 5, |
| "category": self.driver.disk_categories.CLOUD, |
| "disk_name": "data1", |
| "description": "description", |
| "device": "/dev/xvdb", |
| "delete_with_instance": True, |
| } |
| node = self.driver.create_node( |
| name=self.name, |
| image=self.fake_image, |
| size=self.fake_size, |
| ex_security_group_id="sg-28ou0f3xa", |
| ex_data_disks=self.data_disk, |
| ) |
| self.assertIsNotNone(node) |
| |
| def test_list_sizes(self): |
| sizes = self.driver.list_sizes() |
| self.assertEqual(2, len(sizes)) |
| size = sizes[0] |
| self.assertEqual("ecs.t1.xsmall", size.id) |
| self.assertEqual("ecs.t1.xsmall", size.name) |
| self.assertEqual(0.5, size.ram) |
| self.assertEqual(1, size.extra["cpu_core_count"]) |
| self.assertEqual("ecs.t1", size.extra["instance_type_family"]) |
| size = sizes[1] |
| self.assertEqual("ecs.s2.small", size.id) |
| self.assertEqual("ecs.s2.small", size.name) |
| self.assertEqual(1.0, size.ram) |
| self.assertEqual(2, size.extra["cpu_core_count"]) |
| self.assertEqual("ecs.s2", size.extra["instance_type_family"]) |
| |
| def test_list_locations(self): |
| locations = self.driver.list_locations() |
| self.assertEqual(9, len(locations)) |
| location = locations[0] |
| self.assertEqual("ap-southeast-1", location.id) |
| self.assertIsNone(location.country) |
| |
| def test_create_node_without_sg_id_exception(self): |
| name = "test_create_node_without_sg_id_exception" |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_node, |
| name=name, |
| image=self.fake_image, |
| size=self.fake_size, |
| ) |
| |
| def test_creat_node_paybytraffic_exception(self): |
| name = "test_create_node_paybytraffic_exception" |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_node, |
| name=name, |
| image=self.fake_image, |
| size=self.fake_size, |
| ex_security_group_id="sg-id1", |
| ex_internet_charge_type="PayByTraffic", |
| ) |
| |
| def test_create_node_ex_system_disk_exception(self): |
| name = "test_creat_node_ex_system_disk_exception" |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_node, |
| name=name, |
| image=self.fake_image, |
| size=self.fake_size, |
| ex_security_group_id="sg-id1", |
| ex_system_disk=None, |
| ) |
| |
| def test_create_node_ex_private_ip_address_exception(self): |
| name = "test_create_node_ex_private_ip_address_exception" |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_node, |
| name=name, |
| image=self.fake_image, |
| size=self.fake_size, |
| ex_security_group_id="sg-id1", |
| ex_private_ip_address="1.1.1.2", |
| ) |
| |
| def test_reboot_node(self): |
| ECSMockHttp.type = "reboot_node" |
| result = self.driver.reboot_node(self.fake_node) |
| self.assertTrue(result) |
| |
| def test_reboot_node_with_ex_force_stop(self): |
| ECSMockHttp.type = "reboot_node_force_stop" |
| result = self.driver.reboot_node(self.fake_node, ex_force_stop=True) |
| self.assertTrue(result) |
| |
| def test_destroy_node(self): |
| ECSMockHttp.type = "destroy_node" |
| result = self.driver.destroy_node(self.fake_node) |
| self.assertTrue(result) |
| |
| def test_ex_start_node(self): |
| ECSMockHttp.type = "start_node" |
| result = self.driver.ex_start_node(self.fake_node) |
| self.assertTrue(result) |
| |
| def test_ex_stop_node(self): |
| ECSMockHttp.type = "stop_node" |
| result = self.driver.ex_stop_node(self.fake_node) |
| self.assertTrue(result) |
| |
| def test_stop_node_with_ex_force_stop(self): |
| ECSMockHttp.type = "stop_node_force_stop" |
| result = self.driver.ex_stop_node(self.fake_node, ex_force_stop=True) |
| self.assertTrue(result) |
| |
| def test_create_public_ip(self): |
| ECSMockHttp.type = "create_public_ip" |
| result = self.driver.create_public_ip(self.fake_instance_id) |
| self.assertTrue(result) |
| |
| def test_list_volumes(self): |
| volumes = self.driver.list_volumes() |
| self.assertEqual(2, len(volumes)) |
| volume = volumes[0] |
| self.assertEqual("d-28m5zbua0", volume.id) |
| self.assertEqual("", volume.name) |
| self.assertEqual(5, volume.size) |
| self.assertEqual(StorageVolumeState.AVAILABLE, volume.state) |
| expected_extras = { |
| "region_id": "cn-qingdao", |
| "zone_id": "cn-qingdao-b", |
| "description": "", |
| "type": "data", |
| "category": "cloud", |
| "image_id": "", |
| "source_snapshot_id": "", |
| "product_code": "", |
| "portable": True, |
| "instance_id": "", |
| "device": "", |
| "delete_with_instance": False, |
| "enable_auto_snapshot": False, |
| "creation_time": "2014-07-23T02:44:07Z", |
| "attached_time": "2014-07-23T07:47:35Z", |
| "detached_time": "2014-07-23T08:28:48Z", |
| "disk_charge_type": "PostPaid", |
| "operation_locks": {"lock_reason": None}, |
| } |
| self._validate_extras(expected_extras, volume.extra) |
| volume = volumes[1] |
| self.assertEqual("d-28zfrmo13", volume.id) |
| self.assertEqual("ubuntu1404sys", volume.name) |
| self.assertEqual(5, volume.size) |
| self.assertEqual(StorageVolumeState.INUSE, volume.state) |
| expected_extras = { |
| "region_id": "cn-qingdao", |
| "zone_id": "cn-qingdao-b", |
| "description": "Description", |
| "type": "system", |
| "category": "cloud", |
| "image_id": "ubuntu1404_64_20G_aliaegis_20150325.vhd", |
| "source_snapshot_id": "", |
| "product_code": "", |
| "portable": False, |
| "instance_id": "i-28whl2nj2", |
| "device": "/dev/xvda", |
| "delete_with_instance": True, |
| "enable_auto_snapshot": True, |
| "creation_time": "2014-07-23T02:44:06Z", |
| "attached_time": "2016-01-04T15:02:17Z", |
| "detached_time": "", |
| "disk_charge_type": "PostPaid", |
| "operation_locks": {"lock_reason": None}, |
| } |
| self._validate_extras(expected_extras, volume.extra) |
| |
| def test_list_volumes_with_ex_volume_ids(self): |
| ECSMockHttp.type = "list_volumes_ex_volume_ids" |
| volumes = self.driver.list_volumes(ex_volume_ids=["i-28n7dkvov", "not-existed-id"]) |
| self.assertIsNotNone(volumes) |
| |
| def test_list_volumes_with_ex_filters(self): |
| ECSMockHttp.type = "list_volumes_ex_filters" |
| ex_filters = {"InstanceId": self.fake_node.id} |
| volumes = self.driver.list_volumes(ex_filters=ex_filters) |
| self.assertIsNotNone(volumes) |
| |
| def test_list_volume_snapshots(self): |
| snapshots = self.driver.list_volume_snapshots(self.fake_volume) |
| self.assertEqual(1, len(snapshots)) |
| |
| def test_list_volume_snapshots_with_ex_snapshot_ids(self): |
| ECSMockHttp.type = "list_volume_snapshots_ex_snapshot_ids" |
| ex_snapshot_ids = ["fake-snapshot1"] |
| self.driver.list_volume_snapshots(self.fake_volume, ex_snapshot_ids=ex_snapshot_ids) |
| |
| def test_list_volume_snapshots_with_ex_filters(self): |
| ECSMockHttp.type = "list_volume_snapshots_ex_filters" |
| ex_filters = {"InstanceId": self.fake_node.id} |
| self.driver.list_volume_snapshots(self.fake_volume, ex_filters=ex_filters) |
| |
| def test_create_volume(self): |
| ECSMockHttp.type = "create_volume" |
| self.volume_size = 1 |
| self.volume_name = "fake-volume-name" |
| self.description = "fake-description" |
| self.disk_category = "system" |
| self.client_token = "client_token" |
| volume = self.driver.create_volume( |
| self.volume_size, |
| self.volume_name, |
| snapshot=self.fake_snapshot, |
| ex_zone_id=self.zone, |
| ex_description=self.description, |
| ex_disk_category=self.disk_category, |
| ex_client_token=self.client_token, |
| ) |
| self.assertIsNotNone(volume) |
| |
| def test_create_volume_without_ex_zone_id_exception(self): |
| self.assertRaises(AttributeError, self.driver.create_volume, 1, "fake-volume-name") |
| |
| def test_create_volume_snapshot(self): |
| ECSMockHttp.type = "create_volume_snapshot" |
| self.snapshot_name = "fake-snapshot1" |
| self.description = "fake-description" |
| self.client_token = "client-token" |
| snapshot = self.driver.create_volume_snapshot( |
| self.fake_volume, |
| name=self.snapshot_name, |
| ex_description=self.description, |
| ex_client_token=self.client_token, |
| ) |
| self.assertIsNotNone(snapshot) |
| |
| def test_attach_volume(self): |
| self.device = "/dev/sdb" |
| self.delete_with_instance = True |
| attached = self.driver.attach_volume( |
| self.fake_node, |
| self.fake_volume, |
| device=self.device, |
| ex_delete_with_instance=self.delete_with_instance, |
| ) |
| self.assertTrue(attached) |
| |
| def test_detach_volume(self): |
| self.instance_id = "fake-node1" |
| result = self.driver.detach_volume(self.fake_volume, ex_instance_id=self.instance_id) |
| self.assertTrue(result) |
| |
| def test_detach_volume_query_instance_id(self): |
| ECSMockHttp.type = "detach_volume" |
| result = self.driver.detach_volume(self.fake_volume) |
| self.assertTrue(result) |
| |
| def test_detach_volume_query_instance_id_exception(self): |
| self.assertRaises(AttributeError, self.driver.detach_volume, self.fake_volume) |
| |
| def test_destroy_volume(self): |
| ECSMockHttp.type = "destroy_volume" |
| result = self.driver.destroy_volume(self.fake_volume) |
| self.assertTrue(result) |
| |
| def test_destroy_volume_query_volumes_exception(self): |
| self.assertRaises(LibcloudError, self.driver.destroy_volume, self.fake_volume) |
| |
| def test_destroy_volume_state_exception(self): |
| ECSMockHttp.type = "destroy_volume_state" |
| self.assertRaises(LibcloudError, self.driver.destroy_volume, self.fake_volume) |
| |
| def test_destroy_volume_snapshot(self): |
| result = self.driver.destroy_volume_snapshot(self.fake_snapshot) |
| self.assertTrue(result) |
| |
| def test_destroy_volume_snapshot_exception(self): |
| self.assertRaises(AttributeError, self.driver.destroy_volume_snapshot, self.fake_volume) |
| |
| def test_list_images(self): |
| images = self.driver.list_images(self.fake_location) |
| self.assertEqual(1, len(images)) |
| image = images[0] |
| self.assertEqual("freebsd1001_64_20G_aliaegis_20150527.vhd", image.id) |
| self.assertEqual("freebsd1001_64_20G_aliaegis_20150527.vhd", image.name) |
| expected_extra = { |
| "image_version": "1.0.0", |
| "os_type": "linux", |
| "platform": "Freebsd", |
| "architecture": "x86_64", |
| "description": "freebsd1001_64_20G_aliaegis_20150527.vhd", |
| "size": 20, |
| "image_owner_alias": "system", |
| "os_name": "FreeBSD 10.1 64位", |
| "product_code": "", |
| "is_subscribed": False, |
| "progress": "100%", |
| "creation_time": "2015-06-19T07:25:42Z", |
| "usage": "instance", |
| "is_copied": False, |
| } |
| self._validate_extras(expected_extra, image.extra) |
| expected_dev_mappings = { |
| "snapshot_id": "", |
| "size": 20, |
| "device": "/dev/xvda", |
| "format": "", |
| "import_oss_bucket": "", |
| "import_oss_object": "", |
| } |
| self._validate_extras(expected_dev_mappings, image.extra["disk_device_mappings"]) |
| |
| def test_list_images_with_ex_image_ids(self): |
| ECSMockHttp.type = "list_images_ex_image_ids" |
| self.driver.list_images( |
| location=self.fake_location, |
| ex_image_ids=[self.fake_image.id, "not-existed"], |
| ) |
| |
| def test_list_images_with_ex_image_ids_type_exception(self): |
| self.assertRaises( |
| AttributeError, |
| self.driver.list_images, |
| location=self.fake_location, |
| ex_image_ids={"image_ids": "id1,id2"}, |
| ) |
| |
| def test_list_images_with_ex_filters(self): |
| ECSMockHttp.type = "list_images_ex_filters" |
| ex_filters = {"Status": "Available"} |
| self.driver.list_images(location=self.fake_location, ex_filters=ex_filters) |
| |
| def test_list_images_multiple_pages(self): |
| ECSMockHttp.type = "list_images_pages" |
| images = self.driver.list_images() |
| self.assertEqual(2, len(images)) |
| |
| def test_create_image(self): |
| self.image_name = "fake-image1" |
| self.description = "description" |
| self.image_version = "1.0.0" |
| self.client_token = "client_token" |
| image = self.driver.create_image( |
| None, |
| self.image_name, |
| self.description, |
| ex_snapshot_id=self.fake_snapshot.id, |
| ex_image_version=self.image_version, |
| ex_client_token=self.client_token, |
| ) |
| self.assertIsNotNone(image) |
| |
| def test_creaet_image_exception(self): |
| self.assertRaises(AttributeError, self.driver.create_image, None, None) |
| |
| def test_delete_image(self): |
| result = self.driver.delete_image(self.fake_image) |
| self.assertTrue(result) |
| |
| def test_get_image(self): |
| ECSMockHttp.type = "get_image" |
| image = self.driver.get_image(self.fake_image.id) |
| self.assertIsNotNone(image) |
| |
| def test_get_image_not_found_exception(self): |
| ECSMockHttp.type = "get_image_not_found" |
| self.assertRaises(LibcloudError, self.driver.get_image, self.fake_image.id) |
| |
| def test_copy_image(self): |
| self.image_name = "copied-image1" |
| self.description = "description" |
| self.dest_region = "cn-hangzhou" |
| self.client_token = "client-token" |
| image = self.driver.copy_image( |
| self.region, |
| self.fake_image, |
| self.image_name, |
| description=self.description, |
| ex_destination_region_id=self.dest_region, |
| ex_client_token=self.client_token, |
| ) |
| self.assertIsNotNone(image) |
| |
| def test_copy_image_in_the_same_region(self): |
| ECSMockHttp.type = "copy_image_same_region" |
| image = self.driver.copy_image(self.region, self.fake_image, None) |
| self.assertIsNotNone(image) |
| |
| def test_ex_create_security_group(self): |
| self.sg_description = "description" |
| self.client_token = "client-token" |
| sg_id = self.driver.ex_create_security_group( |
| description=self.sg_description, client_token=self.client_token |
| ) |
| self.assertEqual("sg-F876FF7BA", sg_id) |
| |
| def test_ex_list_security_groups(self): |
| sgs = self.driver.ex_list_security_groups() |
| self.assertEqual(1, len(sgs)) |
| sg = sgs[0] |
| self.assertEqual("sg-28ou0f3xa", sg.id) |
| self.assertEqual("sg-28ou0f3xa", sg.name) |
| self.assertEqual("System created security group.", sg.description) |
| self.assertEqual("", sg.vpc_id) |
| self.assertEqual("2015-06-26T08:35:30Z", sg.creation_time) |
| |
| def test_ex_join_security_group(self): |
| result = self.driver.ex_join_security_group( |
| self.fake_node, group_id=self.fake_security_group_id |
| ) |
| self.assertTrue(result) |
| |
| def test_ex_leave_security_group(self): |
| result = self.driver.ex_leave_security_group( |
| self.fake_node, group_id=self.fake_security_group_id |
| ) |
| self.assertTrue(result) |
| |
| def test_ex_delete_security_group_by_id(self): |
| result = self.driver.ex_delete_security_group_by_id(group_id=self.fake_security_group_id) |
| self.assertTrue(result) |
| |
| def test_ex_modify_security_group_by_id(self): |
| self.sg_name = "name" |
| self.sg_description = "description" |
| result = self.driver.ex_modify_security_group_by_id( |
| group_id=self.fake_security_group_id, |
| name=self.sg_name, |
| description=self.sg_description, |
| ) |
| self.assertTrue(result) |
| |
| def test_ex_list_security_groups_with_ex_filters(self): |
| ECSMockHttp.type = "list_sgs_filters" |
| self.vpc_id = "vpc1" |
| ex_filters = {"VpcId": self.vpc_id} |
| sgs = self.driver.ex_list_security_groups(ex_filters=ex_filters) |
| self.assertEqual(1, len(sgs)) |
| |
| def test_ex_list_security_group_attributes(self): |
| self.sga_nictype = "internet" |
| sgas = self.driver.ex_list_security_group_attributes( |
| group_id=self.fake_security_group_id, nic_type=self.sga_nictype |
| ) |
| self.assertEqual(1, len(sgas)) |
| sga = sgas[0] |
| self.assertEqual("ALL", sga.ip_protocol) |
| self.assertEqual("-1/-1", sga.port_range) |
| self.assertEqual("Accept", sga.policy) |
| self.assertEqual("internet", sga.nic_type) |
| |
| def test_ex_list_zones(self): |
| zones = self.driver.ex_list_zones() |
| self.assertEqual(1, len(zones)) |
| zone = zones[0] |
| self.assertEqual("cn-qingdao-b", zone.id) |
| self.assertEqual(self.driver, zone.driver) |
| self.assertEqual("青岛可用区B", zone.name) |
| self.assertIsNotNone(zone.available_resource_types) |
| self.assertEqual("IoOptimized", zone.available_resource_types[0]) |
| self.assertIsNotNone(zone.available_instance_types) |
| self.assertEqual("ecs.m2.medium", zone.available_instance_types[0]) |
| self.assertIsNotNone(zone.available_disk_categories) |
| self.assertEqual("cloud_ssd", zone.available_disk_categories[0]) |
| |
| |
| class ECSMockHttp(MockHttp): |
| fixtures = ComputeFileFixtures("ecs") |
| |
| def _DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_nodes_ex_node_ids_DescribeInstances(self, method, url, body, headers): |
| params = {"InstanceIds": '["i-28n7dkvov", "not-existed-id"]'} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeInstances(method, url, body, headers) |
| |
| def _list_nodes_ex_filters_DescribeInstances(self, method, url, body, headers): |
| params = {"ZoneId": self.test.zone} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeInstances(method, url, body, headers) |
| |
| def _DescribeInstanceTypes(self, method, url, body, headers): |
| resp_body = self.fixtures.load("describe_instance_types.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeRegions(self, method, url, body, headers): |
| resp_body = self.fixtures.load("describe_regions.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_node_CreateInstance(self, method, url, body, headers): |
| params = { |
| "SecurityGroupId": "sg-28ou0f3xa", |
| "Description": "description", |
| "InternetChargeType": "PayByTraffic", |
| "InternetMaxBandwidthOut": "1", |
| "InternetMaxBandwidthIn": "200", |
| "HostName": "hostname", |
| "Password": "password", |
| "IoOptimized": "optimized", |
| "SystemDisk.Category": "cloud", |
| "SystemDisk.DiskName": "root", |
| "SystemDisk.Description": "sys", |
| "VSwitchId": "vswitch-id1", |
| "PrivateIpAddress": "1.1.1.2", |
| "ClientToken": "client_token", |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("create_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_node_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("create_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_node_StartInstance(self, method, url, body, headers): |
| resp_body = self.fixtures.load("start_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_node_with_data_CreateInstance(self, method, url, body, headers): |
| params = { |
| "SecurityGroupId": "sg-28ou0f3xa", |
| "DataDisk.1.Size": "5", |
| "DataDisk.1.Category": "cloud", |
| "DataDisk.1.DiskName": "data1", |
| "DataDisk.1.Description": "description", |
| "DataDisk.1.Device": "/dev/xvdb", |
| "DataDisk.1.DeleteWithInstance": "true", |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("create_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_node_with_data_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("create_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_node_with_data_StartInstance(self, method, url, body, headers): |
| resp_body = self.fixtures.load("start_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _reboot_node_RebootInstance(self, method, url, body, headers): |
| node_id = self.test.fake_node.id |
| self.assertUrlContainsQueryParams(url, {"InstanceId": node_id, "ForceStop": "false"}) |
| resp_body = self.fixtures.load("reboot_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _reboot_node_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("reboot_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _reboot_node_force_stop_RebootInstance(self, method, url, body, headers): |
| node_id = self.test.fake_node.id |
| self.assertUrlContainsQueryParams(url, {"InstanceId": node_id, "ForceStop": "true"}) |
| resp_body = self.fixtures.load("reboot_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _reboot_node_force_stop_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("reboot_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _destroy_node_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("destroy_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _destroy_node_DeleteInstance(self, method, url, body, headers): |
| node_id = self.test.fake_node.id |
| self.assertUrlContainsQueryParams(url, {"InstanceId": node_id}) |
| resp_body = self.fixtures.load("delete_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _start_node_StartInstance(self, method, url, body, headers): |
| node_id = self.test.fake_node.id |
| self.assertUrlContainsQueryParams(url, {"InstanceId": node_id}) |
| resp_body = self.fixtures.load("start_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _start_node_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("reboot_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _stop_node_StopInstance(self, method, url, body, headers): |
| node_id = self.test.fake_node.id |
| self.assertUrlContainsQueryParams(url, {"InstanceId": node_id, "ForceStop": "false"}) |
| resp_body = self.fixtures.load("stop_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _stop_node_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("stop_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _stop_node_force_stop_StopInstance(self, method, url, body, headers): |
| node_id = self.test.fake_node.id |
| self.assertUrlContainsQueryParams(url, {"InstanceId": node_id, "ForceStop": "true"}) |
| resp_body = self.fixtures.load("stop_instance.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _stop_node_force_stop_DescribeInstances(self, method, url, body, headers): |
| resp_body = self.fixtures.load("stop_node_describe_instances.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeDisks(self, method, url, body, headers): |
| resp_body = self.fixtures.load("describe_disks.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_volumes_ex_volume_ids_DescribeDisks(self, method, url, body, headers): |
| region = self.test.region |
| params = {"DiskIds": '["i-28n7dkvov", "not-existed-id"]', "RegionId": region} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeInstances(method, url, body, headers) |
| |
| def _list_volumes_ex_filters_DescribeDisks(self, method, url, body, headers): |
| params = {"InstanceId": self.test.fake_node.id} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeDisks(method, url, body, headers) |
| |
| def _DescribeSnapshots(self, method, url, body, headers): |
| region = self.test.region |
| volume_id = self.test.fake_volume.id |
| params = {"RegionId": region, "DiskId": volume_id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("describe_snapshots.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_volume_snapshots_ex_snapshot_ids_DescribeSnapshots(self, method, url, body, headers): |
| params = {"RegionId": self.test.region, "SnapshotIds": '["fake-snapshot1"]'} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeSnapshots(method, url, body, headers) |
| |
| def _list_volume_snapshots_ex_filters_DescribeSnapshots(self, method, url, body, headers): |
| params = {"InstanceId": self.test.fake_node.id} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeSnapshots(method, url, body, headers) |
| |
| def _create_volume_CreateDisk(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "DiskName": self.test.volume_name, |
| "Size": str(self.test.volume_size), |
| "ZoneId": self.test.zone, |
| "SnapshotId": self.test.fake_snapshot.id, |
| "Description": self.test.description, |
| "DiskCategory": self.test.disk_category, |
| "ClientToken": self.test.client_token, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("create_disk.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_volume_DescribeDisks(self, method, url, body, headers): |
| resp_body = self.fixtures.load("create_volume_describe_disks.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_volume_snapshot_CreateSnapshot(self, method, url, body, headers): |
| params = { |
| "DiskId": self.test.fake_volume.id, |
| "SnapshotName": self.test.snapshot_name, |
| "Description": self.test.description, |
| "ClientToken": self.test.client_token, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("create_snapshot.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_volume_snapshot_DescribeSnapshots(self, method, url, body, headers): |
| resp_body = self.fixtures.load("describe_snapshots.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _AttachDisk(self, method, url, body, headers): |
| delete_with_instance = str(self.test.delete_with_instance).lower() |
| params = { |
| "InstanceId": self.test.fake_node.id, |
| "DiskId": self.test.fake_volume.id, |
| "Device": self.test.device, |
| "DeleteWithInstance": delete_with_instance, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("attach_disk.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DetachDisk(self, method, url, body, headers): |
| params = { |
| "DiskId": self.test.fake_volume.id, |
| "InstanceId": self.test.instance_id, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("detach_disk.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _detach_volume_DescribeDisks(self, method, url, body, headers): |
| params = {"DiskIds": '["' + self.test.fake_volume.id + '"]'} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("detach_volume_describe_disks.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _detach_volume_DetachDisk(self, method, url, body, headers): |
| params = {"DiskId": self.test.fake_volume.id, "InstanceId": "i-28whl2nj2"} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("detach_disk.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _destroy_volume_DescribeDisks(self, method, url, body, headers): |
| params = {"DiskIds": '["' + self.test.fake_volume.id + '"]'} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("destroy_volume_describe_disks.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _destroy_volume_DeleteDisk(self, method, url, body, headers): |
| params = {"DiskId": self.test.fake_volume.id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("delete_disk.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _destroy_volume_state_DescribeDisks(self, method, url, body, headers): |
| return self._detach_volume_DescribeDisks(method, url, body, headers) |
| |
| def _DeleteSnapshot(self, method, url, body, header): |
| params = {"SnapshotId": self.test.fake_snapshot.id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("delete_snapshot.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeImages(self, method, url, body, headers): |
| params = {"RegionId": self.test.fake_location.id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("describe_images.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_images_pages_DescribeImages(self, method, url, body, headers): |
| if "PageNumber=2" in url: |
| resp_body = self.fixtures.load("pages_describe_images_page2.xml") |
| else: |
| resp_body = self.fixtures.load("pages_describe_images.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_images_ex_image_ids_DescribeImages(self, method, url, body, headers): |
| params = {"ImageId": self.test.fake_image.id + ",not-existed"} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeImages(method, url, body, headers) |
| |
| def _list_images_ex_filters_DescribeImages(self, method, url, body, headers): |
| params = {"Status": "Available"} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeImages(method, url, body, headers) |
| |
| def _CreateImage(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "ImageName": self.test.image_name, |
| "Description": self.test.description, |
| "SnapshotId": self.test.fake_snapshot.id, |
| "ImageVersion": self.test.image_version, |
| "ClientToken": self.test.client_token, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("create_image.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteImage(self, method, url, body, headers): |
| params = {"RegionId": self.test.region, "ImageId": self.test.fake_image.id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("delete_image.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _get_image_DescribeImages(self, method, url, body, headers): |
| params = {"RegionId": self.test.region, "ImageId": self.test.fake_image.id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("describe_images.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _get_image_not_found_DescribeImages(self, method, url, body, headers): |
| params = {"RegionId": self.test.region, "ImageId": self.test.fake_image.id} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("get_image_describe_images.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _CopyImage(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "ImageId": self.test.fake_image.id, |
| "DestinationRegionId": self.test.dest_region, |
| "DestinationImageName": self.test.image_name, |
| "DestinationDescription": self.test.description, |
| "ClientToken": self.test.client_token, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("copy_image.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _copy_image_same_region_CopyImage(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "ImageId": self.test.fake_image.id, |
| "DestinationRegionId": self.test.region, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("copy_image.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _copy_image_same_region_DescribeImages(self, method, url, body, headers): |
| return self._DescribeImages(method, url, body, headers) |
| |
| def _DescribeSecurityGroups(self, method, url, body, headers): |
| params = {"RegionId": self.test.region} |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("describe_security_groups.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _JoinSecurityGroup(self, method, url, body, headers): |
| params = { |
| "InstanceId": self.test.fake_node.id, |
| "SecurityGroupId": self.test.fake_security_group_id, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("join_security_group_by_id.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _LeaveSecurityGroup(self, method, url, body, headers): |
| params = { |
| "InstanceId": self.test.fake_node.id, |
| "SecurityGroupId": self.test.fake_security_group_id, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("leave_security_group_by_id.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_sgs_filters_DescribeSecurityGroups(self, method, url, body, headers): |
| params = {"VpcId": self.test.vpc_id} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeSecurityGroups(method, url, body, headers) |
| |
| def _CreateSecurityGroup(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "Description": self.test.sg_description, |
| "ClientToken": self.test.client_token, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("create_security_group.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteSecurityGroup(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "SecurityGroupId": self.test.fake_security_group_id, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("delete_security_group_by_id.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _ModifySecurityGroupAttribute(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "SecurityGroupId": self.test.fake_security_group_id, |
| "SecurityGroupName": self.test.sg_name, |
| "Description": self.test.sg_description, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("modify_security_group_by_id.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeSecurityGroupAttribute(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "SecurityGroupId": self.test.fake_security_group_id, |
| "NicType": self.test.sga_nictype, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| resp_body = self.fixtures.load("describe_security_group_attributes.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeZones(self, method, url, body, headers): |
| resp_body = self.fixtures.load("describe_zones.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_public_ip_AllocatePublicIpAddress(self, method, url, body, headers): |
| resp_body = self.fixtures.load("create_public_ip.xml") |
| return (httplib.OK, resp_body, {}, httplib.responses[httplib.OK]) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
