You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@libcloud.apache.org by to...@apache.org on 2011/09/07 13:58:17 UTC
svn commit: r1166136 [2/2] - in /libcloud/trunk: libcloud/ libcloud/common/
libcloud/compute/drivers/ libcloud/loadbalancer/drivers/
libcloud/storage/drivers/ test/ test/compute/
test/compute/fixtures/openstack/ test/compute/fixtures/rackspace/
Added: libcloud/trunk/test/compute/test_openstack.py
URL: http://svn.apache.org/viewvc/libcloud/trunk/test/compute/test_openstack.py?rev=1166136&view=auto
==============================================================================
--- libcloud/trunk/test/compute/test_openstack.py (added)
+++ libcloud/trunk/test/compute/test_openstack.py Wed Sep 7 11:58:16 2011
@@ -0,0 +1,420 @@
+# 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
+import httplib
+
+from libcloud.common.types import InvalidCredsError, MalformedResponseError
+from libcloud.compute.drivers.rackspace import OpenStackNodeDriver, OpenStackResponse
+from libcloud.compute.base import Node, NodeImage, NodeSize
+from libcloud.pricing import set_pricing, clear_pricing_data
+
+from test import MockResponse, MockHttpTestCase, XML_HEADERS
+from test.file_fixtures import ComputeFileFixtures
+from test.compute import TestCaseMixin
+
+from test.secrets import NOVA_USERNAME, NOVA_API_KEY, NOVA_HOST, NOVA_PORT
+from test.secrets import NOVA_SECURE
+
+
+class OpenStackResponseTestCase(unittest.TestCase):
+ XML = """<?xml version="1.0" encoding="UTF-8"?><root/>"""
+
+ def test_simple_xml_content_type_handling(self):
+ http_response = MockResponse(200, OpenStackResponseTestCase.XML, headers={'content-type': 'application/xml'})
+ body = OpenStackResponse(http_response).parse_body()
+
+ self.assertTrue(hasattr(body, 'tag'), "Body should be parsed as XML")
+
+ def test_extended_xml_content_type_handling(self):
+ http_response = MockResponse(200,
+ OpenStackResponseTestCase.XML,
+ headers={'content-type': 'application/xml; charset=UTF-8'})
+ body = OpenStackResponse(http_response).parse_body()
+
+ self.assertTrue(hasattr(body, 'tag'), "Body should be parsed as XML")
+
+ def test_non_xml_content_type_handling(self):
+ RESPONSE_BODY = "Accepted"
+
+ http_response = MockResponse(202, RESPONSE_BODY, headers={'content-type': 'text/html'})
+ body = OpenStackResponse(http_response).parse_body()
+
+ self.assertEqual(body, RESPONSE_BODY, "Non-XML body should be returned as is")
+
+
+class OpenStackTests(unittest.TestCase, TestCaseMixin):
+ should_list_locations = False
+
+ driver_type = OpenStackNodeDriver
+ driver_args = (
+ NOVA_USERNAME,
+ NOVA_API_KEY,
+ NOVA_SECURE,
+ NOVA_HOST,
+ NOVA_PORT,
+ )
+
+ @classmethod
+ def create_driver(self):
+ return self.driver_type(*self.driver_args)
+
+ def setUp(self):
+ self.driver_type.connectionCls.conn_classes = (OpenStackMockHttp, OpenStackMockHttp)
+ OpenStackMockHttp.type = None
+ self.driver = self.create_driver()
+ clear_pricing_data()
+
+ def test_auth(self):
+ OpenStackMockHttp.type = 'UNAUTHORIZED'
+ try:
+ self.driver = self.create_driver()
+ except InvalidCredsError, e:
+ self.assertEqual(True, isinstance(e, InvalidCredsError))
+ else:
+ self.fail('test should have thrown')
+
+ def test_auth_missing_key(self):
+ OpenStackMockHttp.type = 'UNAUTHORIZED_MISSING_KEY'
+ try:
+ self.driver = self.create_driver()
+ except MalformedResponseError, e:
+ self.assertEqual(True, isinstance(e, MalformedResponseError))
+ else:
+ self.fail('test should have thrown')
+
+ def test_auth_server_error(self):
+ OpenStackMockHttp.type = 'INTERNAL_SERVER_ERROR'
+ try:
+ self.driver = self.create_driver()
+ except MalformedResponseError, e:
+ self.assertEqual(True, isinstance(e, MalformedResponseError))
+ else:
+ self.fail('test should have thrown')
+
+ def test_list_nodes(self):
+ OpenStackMockHttp.type = 'EMPTY'
+ ret = self.driver.list_nodes()
+ self.assertEqual(len(ret), 0)
+ OpenStackMockHttp.type = None
+ ret = self.driver.list_nodes()
+ self.assertEqual(len(ret), 1)
+ node = ret[0]
+ self.assertEqual('67.23.21.33', node.public_ip[0])
+ self.assertEqual('10.176.168.218', node.private_ip[0])
+ self.assertEqual(node.extra.get('flavorId'), '1')
+ self.assertEqual(node.extra.get('imageId'), '11')
+ self.assertEqual(type(node.extra.get('metadata')), type(dict()))
+ OpenStackMockHttp.type = 'METADATA'
+ ret = self.driver.list_nodes()
+ self.assertEqual(len(ret), 1)
+ node = ret[0]
+ self.assertEqual(type(node.extra.get('metadata')), type(dict()))
+ self.assertEqual(node.extra.get('metadata').get('somekey'),
+ 'somevalue')
+ OpenStackMockHttp.type = None
+
+ def test_list_images(self):
+ ret = self.driver.list_images()
+ expected = {10: {'serverId': None,
+ 'status': 'ACTIVE',
+ 'created': '2009-07-20T09:14:37-05:00',
+ 'updated': '2009-07-20T09:14:37-05:00',
+ 'progress': None},
+ 11: {'serverId': '91221',
+ 'status': 'ACTIVE',
+ 'created': '2009-11-29T20:22:09-06:00',
+ 'updated': '2009-11-29T20:24:08-06:00',
+ 'progress': '100'}}
+ for ret_idx, extra in expected.items():
+ for key, value in extra.items():
+ self.assertEqual(ret[ret_idx].extra[key], value)
+
+ def test_create_node(self):
+ image = NodeImage(id=11, name='Ubuntu 8.10 (intrepid)',
+ driver=self.driver)
+ size = NodeSize(1, '256 slice', None, None, None, None,
+ driver=self.driver)
+ node = self.driver.create_node(name='racktest', image=image, size=size)
+ self.assertEqual(node.name, 'racktest')
+ self.assertEqual(node.extra.get('password'), 'racktestvJq7d3')
+
+ def test_create_node_ex_shared_ip_group(self):
+ OpenStackMockHttp.type = 'EX_SHARED_IP_GROUP'
+ image = NodeImage(id=11, name='Ubuntu 8.10 (intrepid)',
+ driver=self.driver)
+ size = NodeSize(1, '256 slice', None, None, None, None,
+ driver=self.driver)
+ node = self.driver.create_node(name='racktest', image=image, size=size,
+ ex_shared_ip_group_id='12345')
+ self.assertEqual(node.name, 'racktest')
+ self.assertEqual(node.extra.get('password'), 'racktestvJq7d3')
+
+ def test_create_node_with_metadata(self):
+ OpenStackMockHttp.type = 'METADATA'
+ image = NodeImage(id=11, name='Ubuntu 8.10 (intrepid)',
+ driver=self.driver)
+ size = NodeSize(1, '256 slice', None, None, None, None,
+ driver=self.driver)
+ metadata = {'a': 'b', 'c': 'd'}
+ files = {'/file1': 'content1', '/file2': 'content2'}
+ node = self.driver.create_node(name='racktest', image=image, size=size,
+ metadata=metadata, files=files)
+ self.assertEqual(node.name, 'racktest')
+ self.assertEqual(node.extra.get('password'), 'racktestvJq7d3')
+ self.assertEqual(node.extra.get('metadata'), metadata)
+
+ def test_reboot_node(self):
+ node = Node(id=72258, name=None, state=None, public_ip=None,
+ private_ip=None, driver=self.driver)
+ ret = node.reboot()
+ self.assertTrue(ret is True)
+
+ def test_destroy_node(self):
+ node = Node(id=72258, name=None, state=None, public_ip=None,
+ private_ip=None, driver=self.driver)
+ ret = node.destroy()
+ self.assertTrue(ret is True)
+
+ def test_ex_limits(self):
+ limits = self.driver.ex_limits()
+ self.assertTrue("rate" in limits)
+ self.assertTrue("absolute" in limits)
+
+ def test_ex_save_image(self):
+ node = Node(id=444222, name=None, state=None, public_ip=None,
+ private_ip=None, driver=self.driver)
+ image = self.driver.ex_save_image(node, "imgtest")
+ self.assertEqual(image.name, "imgtest")
+ self.assertEqual(image.id, "12345")
+
+ def test_ex_list_ip_addresses(self):
+ ret = self.driver.ex_list_ip_addresses(node_id=72258)
+ self.assertEquals(2, len(ret.public_addresses))
+ self.assertTrue('67.23.10.131' in ret.public_addresses)
+ self.assertTrue('67.23.10.132' in ret.public_addresses)
+ self.assertEquals(1, len(ret.private_addresses))
+ self.assertTrue('10.176.42.16' in ret.private_addresses)
+
+ def test_ex_list_ip_groups(self):
+ ret = self.driver.ex_list_ip_groups()
+ self.assertEquals(2, len(ret))
+ self.assertEquals('1234', ret[0].id)
+ self.assertEquals('Shared IP Group 1', ret[0].name)
+ self.assertEquals('5678', ret[1].id)
+ self.assertEquals('Shared IP Group 2', ret[1].name)
+ self.assertTrue(ret[0].servers is None)
+
+ def test_ex_list_ip_groups_detail(self):
+ ret = self.driver.ex_list_ip_groups(details=True)
+
+ self.assertEquals(2, len(ret))
+
+ self.assertEquals('1234', ret[0].id)
+ self.assertEquals('Shared IP Group 1', ret[0].name)
+ self.assertEquals(2, len(ret[0].servers))
+ self.assertEquals('422', ret[0].servers[0])
+ self.assertEquals('3445', ret[0].servers[1])
+
+ self.assertEquals('5678', ret[1].id)
+ self.assertEquals('Shared IP Group 2', ret[1].name)
+ self.assertEquals(3, len(ret[1].servers))
+ self.assertEquals('23203', ret[1].servers[0])
+ self.assertEquals('2456', ret[1].servers[1])
+ self.assertEquals('9891', ret[1].servers[2])
+
+ def test_ex_create_ip_group(self):
+ ret = self.driver.ex_create_ip_group('Shared IP Group 1', '5467')
+ self.assertEquals('1234', ret.id)
+ self.assertEquals('Shared IP Group 1', ret.name)
+ self.assertEquals(1, len(ret.servers))
+ self.assertEquals('422', ret.servers[0])
+
+ def test_ex_delete_ip_group(self):
+ ret = self.driver.ex_delete_ip_group('5467')
+ self.assertEquals(True, ret)
+
+ def test_ex_share_ip(self):
+ ret = self.driver.ex_share_ip('1234', '3445', '67.23.21.133')
+ self.assertEquals(True, ret)
+
+ def test_ex_unshare_ip(self):
+ ret = self.driver.ex_unshare_ip('3445', '67.23.21.133')
+ self.assertEquals(True, ret)
+
+ def test_ex_resize(self):
+ node = Node(id=444222, name=None, state=None, public_ip=None,
+ private_ip=None, driver=self.driver)
+ size = NodeSize(1, '256 slice', None, None, None, None,
+ driver=self.driver)
+ self.assertTrue(self.driver.ex_resize(node=node, size=size))
+
+ def test_ex_confirm_resize(self):
+ node = Node(id=444222, name=None, state=None, public_ip=None,
+ private_ip=None, driver=self.driver)
+ self.assertTrue(self.driver.ex_confirm_resize(node=node))
+
+ def test_ex_revert_resize(self):
+ node = Node(id=444222, name=None, state=None, public_ip=None,
+ private_ip=None, driver=self.driver)
+ self.assertTrue(self.driver.ex_revert_resize(node=node))
+
+ def test_list_sizes(self):
+ sizes = self.driver.list_sizes()
+ self.assertEqual(len(sizes), 7, 'Wrong sizes count')
+
+ for size in sizes:
+ self.assertTrue(isinstance(size.price, float),
+ 'Wrong size price type')
+ self.assertEqual(size.price, 0,
+ 'Size price should be zero by default')
+
+ def test_list_sizes_with_specified_pricing(self):
+ pricing = dict((str(i), i) for i in range(1, 9))
+
+ set_pricing(driver_type='compute', driver_name='openstack',
+ pricing=pricing)
+
+ sizes = self.driver.list_sizes()
+ self.assertEqual(len(sizes), 7, 'Wrong sizes count')
+
+ for size in sizes:
+ self.assertTrue(isinstance(size.price, float),
+ 'Wrong size price type')
+ self.assertEqual(size.price, pricing[size.id],
+ 'Size price should be zero by default')
+
+
+class OpenStackMockHttp(MockHttpTestCase):
+ fixtures = ComputeFileFixtures('openstack')
+
+ # fake auth token response
+ def _v1_0(self, method, url, body, headers):
+ headers = {'x-server-management-url': 'https://servers.api.rackspacecloud.com/v1.0/slug',
+ 'x-auth-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
+ 'x-cdn-management-url': 'https://cdn.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06',
+ 'x-storage-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
+ 'x-storage-url': 'https://storage4.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06'}
+ return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
+
+ def _v1_0_UNAUTHORIZED(self, method, url, body, headers):
+ return (httplib.UNAUTHORIZED, "", {}, httplib.responses[httplib.UNAUTHORIZED])
+
+ def _v1_0_INTERNAL_SERVER_ERROR(self, method, url, body, headers):
+ return (httplib.INTERNAL_SERVER_ERROR, "<h1>500: Internal Server Error</h1>", {}, httplib.responses[httplib.INTERNAL_SERVER_ERROR])
+
+ def _v1_0_UNAUTHORIZED_MISSING_KEY(self, method, url, body, headers):
+ headers = {'x-server-management-url': 'https://servers.api.rackspacecloud.com/v1.0/slug',
+ 'x-auth-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
+ 'x-cdn-management-url': 'https://cdn.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06'}
+ return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
+
+ def _v1_0_slug_servers_detail_EMPTY(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_servers_detail_empty.xml')
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_servers_detail(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_servers_detail.xml')
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_servers_detail_METADATA(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_servers_detail_metadata.xml')
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_images(self, method, url, body, headers):
+ if method != "POST":
+ raise NotImplemented
+ # this is currently used for creation of new image with
+ # POST request, don't handle GET to avoid possible confusion
+ body = self.fixtures.load('v1_slug_images_post.xml')
+ return (httplib.ACCEPTED, body, XML_HEADERS, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_images_detail(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_images_detail.xml')
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_servers(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_servers.xml')
+ return (httplib.ACCEPTED, body, XML_HEADERS, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_servers_EX_SHARED_IP_GROUP(self, method, url, body, headers):
+ # test_create_node_ex_shared_ip_group
+ # Verify that the body contains sharedIpGroupId XML element
+ self.assertTrue(body.find('sharedIpGroupId="12345"') != -1)
+ body = self.fixtures.load('v1_slug_servers.xml')
+ return (httplib.ACCEPTED, body, XML_HEADERS, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_servers_METADATA(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_servers_metadata.xml')
+ return (httplib.ACCEPTED, body, XML_HEADERS, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_servers_72258_action(self, method, url, body, headers):
+ if method != "POST" or body[:8] != "<reboot ":
+ raise NotImplemented
+ # only used by reboot() right now, but we will need to parse body someday !!!!
+ return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_limits(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_limits.xml')
+ return (httplib.ACCEPTED, body, XML_HEADERS, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_servers_72258(self, method, url, body, headers):
+ if method != "DELETE":
+ raise NotImplemented
+ # only used by destroy node()
+ return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_servers_72258_ips(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_servers_ips.xml')
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_shared_ip_groups_5467(self, method, url, body, headers):
+ if method != 'DELETE':
+ raise NotImplemented
+ return (httplib.NO_CONTENT, "", {}, httplib.responses[httplib.NO_CONTENT])
+
+ def _v1_0_slug_shared_ip_groups(self, method, url, body, headers):
+
+ fixture = 'v1_slug_shared_ip_group.xml' if method == 'POST' else 'v1_slug_shared_ip_groups.xml'
+ body = self.fixtures.load(fixture)
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_shared_ip_groups_detail(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_shared_ip_groups_detail.xml')
+ return (httplib.OK, body, XML_HEADERS, httplib.responses[httplib.OK])
+
+ def _v1_0_slug_servers_3445_ips_public_67_23_21_133(self, method, url, body, headers):
+ return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED])
+
+ def _v1_0_slug_servers_444222_action(self, method, url, body, headers):
+ if body.find('resize') != -1:
+ # test_ex_resize_server
+ return (httplib.ACCEPTED, "", headers, httplib.responses[httplib.NO_CONTENT])
+ elif body.find('confirmResize') != -1:
+ # test_ex_confirm_resize
+ return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
+ elif body.find('revertResize') != -1:
+ # test_ex_revert_resize
+ return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
+
+ def _v1_0_slug_flavors_detail(self, method, url, body, headers):
+ body = self.fixtures.load('v1_slug_flavors_detail.xml')
+ headers = {'date': 'Tue, 14 Jun 2011 09:43:55 GMT', 'content-length': '529'}
+ headers.update(XML_HEADERS)
+ return (httplib.OK, body, headers, httplib.responses[httplib.OK])
+
+
+if __name__ == '__main__':
+ sys.exit(unittest.main())
Modified: libcloud/trunk/test/compute/test_rackspace.py
URL: http://svn.apache.org/viewvc/libcloud/trunk/test/compute/test_rackspace.py?rev=1166136&r1=1166135&r2=1166136&view=diff
==============================================================================
--- libcloud/trunk/test/compute/test_rackspace.py (original)
+++ libcloud/trunk/test/compute/test_rackspace.py Wed Sep 7 11:58:16 2011
@@ -14,450 +14,22 @@
# limitations under the License.
import sys
import unittest
-import httplib
-from libcloud.common.types import InvalidCredsError, MalformedResponseError
-from libcloud.compute.drivers.rackspace import RackspaceNodeDriver as Rackspace
-from libcloud.compute.drivers.rackspace import OpenStackResponse
-from libcloud.compute.drivers.rackspace import OpenStackNodeDriver as OpenStack
-from libcloud.compute.base import Node, NodeImage, NodeSize
-from libcloud.pricing import set_pricing
-
-from test import MockHttp, MockResponse, MockHttpTestCase
-from test.compute import TestCaseMixin
-from test.file_fixtures import ComputeFileFixtures
+from libcloud.compute.drivers.rackspace import RackspaceNodeDriver
+from test.compute.test_openstack import OpenStackTests
from test.secrets import RACKSPACE_USER, RACKSPACE_KEY
-from test.secrets import NOVA_USERNAME, NOVA_API_KEY, NOVA_HOST, NOVA_PORT
-from test.secrets import NOVA_SECURE
-
-
-class RackspaceTests(unittest.TestCase, TestCaseMixin):
-
- def setUp(self):
- Rackspace.connectionCls.conn_classes = (None, RackspaceMockHttp)
- RackspaceMockHttp.type = None
- self.driver = Rackspace(RACKSPACE_USER, RACKSPACE_KEY)
-
- def test_auth(self):
- RackspaceMockHttp.type = 'UNAUTHORIZED'
- try:
- self.driver = Rackspace(RACKSPACE_USER, RACKSPACE_KEY)
- except InvalidCredsError, e:
- self.assertEqual(True, isinstance(e, InvalidCredsError))
- else:
- self.fail('test should have thrown')
-
- def test_auth_missing_key(self):
- RackspaceMockHttp.type = 'UNAUTHORIZED_MISSING_KEY'
- try:
- self.driver = Rackspace(RACKSPACE_USER, RACKSPACE_KEY)
- except MalformedResponseError, e:
- self.assertEqual(True, isinstance(e, MalformedResponseError))
- else:
- self.fail('test should have thrown')
-
- def test_auth_server_error(self):
- RackspaceMockHttp.type = 'INTERNAL_SERVER_ERROR'
- try:
- self.driver = Rackspace(RACKSPACE_USER, RACKSPACE_KEY)
- except MalformedResponseError, e:
- self.assertEqual(True, isinstance(e, MalformedResponseError))
- else:
- self.fail('test should have thrown')
-
- def test_list_nodes(self):
- RackspaceMockHttp.type = 'EMPTY'
- ret = self.driver.list_nodes()
- self.assertEqual(len(ret), 0)
- RackspaceMockHttp.type = None
- ret = self.driver.list_nodes()
- self.assertEqual(len(ret), 1)
- node = ret[0]
- self.assertEqual('67.23.21.33', node.public_ip[0])
- self.assertEqual('10.176.168.218', node.private_ip[0])
- self.assertEqual(node.extra.get('flavorId'), '1')
- self.assertEqual(node.extra.get('imageId'), '11')
- self.assertEqual(type(node.extra.get('metadata')), type(dict()))
- RackspaceMockHttp.type = 'METADATA'
- ret = self.driver.list_nodes()
- self.assertEqual(len(ret), 1)
- node = ret[0]
- self.assertEqual(type(node.extra.get('metadata')), type(dict()))
- self.assertEqual(node.extra.get('metadata').get('somekey'),
- 'somevalue')
- RackspaceMockHttp.type = None
-
- def test_list_sizes(self):
- ret = self.driver.list_sizes()
- self.assertEqual(len(ret), 7)
- size = ret[0]
- self.assertEqual(size.name, '256 slice')
- self.assertTrue(isinstance(size.price, float))
-
- def test_list_images(self):
- ret = self.driver.list_images()
- expected = {10: {'serverId': None,
- 'status': 'ACTIVE',
- 'created': '2009-07-20T09:14:37-05:00',
- 'updated': '2009-07-20T09:14:37-05:00',
- 'progress': None},
- 11: {'serverId': '91221',
- 'status': 'ACTIVE',
- 'created': '2009-11-29T20:22:09-06:00',
- 'updated': '2009-11-29T20:24:08-06:00',
- 'progress': '100'}}
- for ret_idx, extra in expected.items():
- for key, value in extra.items():
- self.assertEqual(ret[ret_idx].extra[key], value)
-
- def test_create_node(self):
- image = NodeImage(id=11, name='Ubuntu 8.10 (intrepid)',
- driver=self.driver)
- size = NodeSize(1, '256 slice', None, None, None, None,
- driver=self.driver)
- node = self.driver.create_node(name='racktest', image=image, size=size)
- self.assertEqual(node.name, 'racktest')
- self.assertEqual(node.extra.get('password'), 'racktestvJq7d3')
-
- def test_create_node_ex_shared_ip_group(self):
- RackspaceMockHttp.type = 'EX_SHARED_IP_GROUP'
- image = NodeImage(id=11, name='Ubuntu 8.10 (intrepid)',
- driver=self.driver)
- size = NodeSize(1, '256 slice', None, None, None, None,
- driver=self.driver)
- node = self.driver.create_node(name='racktest', image=image, size=size,
- ex_shared_ip_group_id='12345')
- self.assertEqual(node.name, 'racktest')
- self.assertEqual(node.extra.get('password'), 'racktestvJq7d3')
-
- def test_create_node_with_metadata(self):
- RackspaceMockHttp.type = 'METADATA'
- image = NodeImage(id=11, name='Ubuntu 8.10 (intrepid)',
- driver=self.driver)
- size = NodeSize(1, '256 slice', None, None, None, None,
- driver=self.driver)
- metadata = {'a': 'b', 'c': 'd'}
- files = {'/file1': 'content1', '/file2': 'content2'}
- node = self.driver.create_node(name='racktest', image=image, size=size,
- metadata=metadata, files=files)
- self.assertEqual(node.name, 'racktest')
- self.assertEqual(node.extra.get('password'), 'racktestvJq7d3')
- self.assertEqual(node.extra.get('metadata'), metadata)
-
- def test_reboot_node(self):
- node = Node(id=72258, name=None, state=None, public_ip=None,
- private_ip=None, driver=self.driver)
- ret = node.reboot()
- self.assertTrue(ret is True)
-
- def test_destroy_node(self):
- node = Node(id=72258, name=None, state=None, public_ip=None,
- private_ip=None, driver=self.driver)
- ret = node.destroy()
- self.assertTrue(ret is True)
-
- def test_ex_limits(self):
- limits = self.driver.ex_limits()
- self.assertTrue("rate" in limits)
- self.assertTrue("absolute" in limits)
-
- def test_ex_save_image(self):
- node = Node(id=444222, name=None, state=None, public_ip=None,
- private_ip=None, driver=self.driver)
- image = self.driver.ex_save_image(node, "imgtest")
- self.assertEqual(image.name, "imgtest")
- self.assertEqual(image.id, "12345")
-
- def test_ex_list_ip_addresses(self):
- ret = self.driver.ex_list_ip_addresses(node_id=72258)
- self.assertEquals(2, len(ret.public_addresses))
- self.assertTrue('67.23.10.131' in ret.public_addresses)
- self.assertTrue('67.23.10.132' in ret.public_addresses)
- self.assertEquals(1, len(ret.private_addresses))
- self.assertTrue('10.176.42.16' in ret.private_addresses)
-
- def test_ex_list_ip_groups(self):
- ret = self.driver.ex_list_ip_groups()
- self.assertEquals(2, len(ret))
- self.assertEquals('1234', ret[0].id)
- self.assertEquals('Shared IP Group 1', ret[0].name)
- self.assertEquals('5678', ret[1].id)
- self.assertEquals('Shared IP Group 2', ret[1].name)
- self.assertTrue(ret[0].servers is None)
-
- def test_ex_list_ip_groups_detail(self):
- ret = self.driver.ex_list_ip_groups(details=True)
-
- self.assertEquals(2, len(ret))
-
- self.assertEquals('1234', ret[0].id)
- self.assertEquals('Shared IP Group 1', ret[0].name)
- self.assertEquals(2, len(ret[0].servers))
- self.assertEquals('422', ret[0].servers[0])
- self.assertEquals('3445', ret[0].servers[1])
-
- self.assertEquals('5678', ret[1].id)
- self.assertEquals('Shared IP Group 2', ret[1].name)
- self.assertEquals(3, len(ret[1].servers))
- self.assertEquals('23203', ret[1].servers[0])
- self.assertEquals('2456', ret[1].servers[1])
- self.assertEquals('9891', ret[1].servers[2])
-
- def test_ex_create_ip_group(self):
- ret = self.driver.ex_create_ip_group('Shared IP Group 1', '5467')
- self.assertEquals('1234', ret.id)
- self.assertEquals('Shared IP Group 1', ret.name)
- self.assertEquals(1, len(ret.servers))
- self.assertEquals('422', ret.servers[0])
-
- def test_ex_delete_ip_group(self):
- ret = self.driver.ex_delete_ip_group('5467')
- self.assertEquals(True, ret)
-
- def test_ex_share_ip(self):
- ret = self.driver.ex_share_ip('1234', '3445', '67.23.21.133')
- self.assertEquals(True, ret)
-
- def test_ex_unshare_ip(self):
- ret = self.driver.ex_unshare_ip('3445', '67.23.21.133')
- self.assertEquals(True, ret)
-
- def test_ex_resize(self):
- node = Node(id=444222, name=None, state=None, public_ip=None,
- private_ip=None, driver=self.driver)
- size = NodeSize(1, '256 slice', None, None, None, None,
- driver=self.driver)
- self.assertTrue(self.driver.ex_resize(node=node, size=size))
-
- def test_ex_confirm_resize(self):
- node = Node(id=444222, name=None, state=None, public_ip=None,
- private_ip=None, driver=self.driver)
- self.assertTrue(self.driver.ex_confirm_resize(node=node))
-
- def test_ex_revert_resize(self):
- node = Node(id=444222, name=None, state=None, public_ip=None,
- private_ip=None, driver=self.driver)
- self.assertTrue(self.driver.ex_revert_resize(node=node))
-
-
-class RackspaceMockHttp(MockHttpTestCase):
-
- fixtures = ComputeFileFixtures('rackspace')
-
- # fake auth token response
- def _v1_0(self, method, url, body, headers):
- headers = {'x-server-management-url': 'https://servers.api.rackspacecloud.com/v1.0/slug',
- 'x-auth-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-cdn-management-url': 'https://cdn.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-storage-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-storage-url': 'https://storage4.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06'}
- return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
-
- def _v1_0_UNAUTHORIZED(self, method, url, body, headers):
- return (httplib.UNAUTHORIZED, "", {}, httplib.responses[httplib.UNAUTHORIZED])
-
- def _v1_0_INTERNAL_SERVER_ERROR(self, method, url, body, headers):
- return (httplib.INTERNAL_SERVER_ERROR, "<h1>500: Internal Server Error</h1>", {}, httplib.responses[httplib.INTERNAL_SERVER_ERROR])
-
- def _v1_0_UNAUTHORIZED_MISSING_KEY(self, method, url, body, headers):
- headers = {'x-server-management-url': 'https://servers.api.rackspacecloud.com/v1.0/slug',
- 'x-auth-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-cdn-management-url': 'https://cdn.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06'}
- return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
-
- def _v1_0_slug_servers_detail_EMPTY(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_servers_detail_empty.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_servers_detail(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_servers_detail.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_servers_detail_METADATA(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_servers_detail_metadata.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_flavors_detail(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_flavors_detail.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_images(self, method, url, body, headers):
- if method != "POST":
- raise NotImplemented
- # this is currently used for creation of new image with
- # POST request, don't handle GET to avoid possible confusion
- body = self.fixtures.load('v1_slug_images_post.xml')
- return (httplib.ACCEPTED, body, {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_images_detail(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_images_detail.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_servers(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_servers.xml')
- return (httplib.ACCEPTED, body, {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_servers_EX_SHARED_IP_GROUP(self, method, url, body, headers):
- # test_create_node_ex_shared_ip_group
- # Verify that the body contains sharedIpGroupId XML element
- self.assertTrue(body.find('sharedIpGroupId="12345"') != -1)
- body = self.fixtures.load('v1_slug_servers.xml')
- return (httplib.ACCEPTED, body, {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_servers_METADATA(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_servers_metadata.xml')
- return (httplib.ACCEPTED, body, {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_servers_72258_action(self, method, url, body, headers):
- if method != "POST" or body[:8] != "<reboot ":
- raise NotImplemented
- # only used by reboot() right now, but we will need to parse body someday !!!!
- return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_limits(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_limits.xml')
- return (httplib.ACCEPTED, body, {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_servers_72258(self, method, url, body, headers):
- if method != "DELETE":
- raise NotImplemented
- # only used by destroy node()
- return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_servers_72258_ips(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_servers_ips.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_shared_ip_groups_5467(self, method, url, body, headers):
- if method != 'DELETE':
- raise NotImplemented
- return (httplib.NO_CONTENT, "", {}, httplib.responses[httplib.NO_CONTENT])
-
- def _v1_0_slug_shared_ip_groups(self, method, url, body, headers):
-
- fixture = 'v1_slug_shared_ip_group.xml' if method == 'POST' else 'v1_slug_shared_ip_groups.xml'
- body = self.fixtures.load(fixture)
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_shared_ip_groups_detail(self, method, url, body, headers):
- body = self.fixtures.load('v1_slug_shared_ip_groups_detail.xml')
- return (httplib.OK, body, {}, httplib.responses[httplib.OK])
-
- def _v1_0_slug_servers_3445_ips_public_67_23_21_133(self, method, url, body, headers):
- return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED])
-
- def _v1_0_slug_servers_444222_action(self, method, url, body, headers):
- if body.find('resize') != -1:
- # test_ex_resize_server
- return (httplib.ACCEPTED, "", headers, httplib.responses[httplib.NO_CONTENT])
- elif body.find('confirmResize') != -1:
- # test_ex_confirm_resize
- return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
- elif body.find('revertResize') != -1:
- # test_ex_revert_resize
- return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
-
-
-class OpenStackResponseTestCase(unittest.TestCase):
- XML = """<?xml version="1.0" encoding="UTF-8"?><root/>"""
-
- def test_simple_xml_content_type_handling(self):
- http_response = MockResponse(200, OpenStackResponseTestCase.XML, headers={'content-type': 'application/xml'})
- body = OpenStackResponse(http_response).parse_body()
-
- self.assertTrue(hasattr(body, 'tag'), "Body should be parsed as XML")
-
- def test_extended_xml_content_type_handling(self):
- http_response = MockResponse(200,
- OpenStackResponseTestCase.XML,
- headers={'content-type': 'application/xml; charset=UTF-8'})
- body = OpenStackResponse(http_response).parse_body()
-
- self.assertTrue(hasattr(body, 'tag'), "Body should be parsed as XML")
-
- def test_non_xml_content_type_handling(self):
- RESPONSE_BODY = "Accepted"
-
- http_response = MockResponse(202, RESPONSE_BODY, headers={'content-type': 'text/html'})
- body = OpenStackResponse(http_response).parse_body()
-
- self.assertEqual(body, RESPONSE_BODY, "Non-XML body should be returned as is")
-
-
-class OpenStackTests(unittest.TestCase):
- def setUp(self):
- OpenStack.connectionCls.conn_classes = (OpenStackMockHttp, None)
- OpenStackMockHttp.type = None
- self.driver = OpenStack(NOVA_USERNAME, NOVA_API_KEY, NOVA_SECURE,
- NOVA_HOST, NOVA_PORT)
-
- def test_destroy_node(self):
- node = Node(id=72258, name=None, state=None, public_ip=None, private_ip=None,
- driver=self.driver)
- ret = node.destroy()
- self.assertTrue(ret is True, 'Unsuccessful node destroying')
-
- def test_list_sizes(self):
- sizes = self.driver.list_sizes()
- self.assertEqual(len(sizes), 8, 'Wrong sizes count')
-
- for size in sizes:
- self.assertTrue(isinstance(size.price, float),
- 'Wrong size price type')
- self.assertEqual(size.price, 0,
- 'Size price should be zero by default')
-
- def test_list_sizes_with_specified_pricing(self):
- pricing = dict((str(i), i) for i in range(1, 9))
-
- set_pricing(driver_type='compute', driver_name='openstack',
- pricing=pricing)
-
- sizes = self.driver.list_sizes()
- self.assertEqual(len(sizes), 8, 'Wrong sizes count')
-
- for size in sizes:
- self.assertTrue(isinstance(size.price, float),
- 'Wrong size price type')
- self.assertEqual(size.price, pricing[size.id],
- 'Size price should be zero by default')
-
-
-class OpenStackMockHttp(MockHttp):
- def _v1_0(self, method, url, body, headers):
- headers = {'x-server-management-url': 'https://servers.api.rackspacecloud.com/v1.0/slug',
- 'x-auth-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-cdn-management-url': 'https://cdn.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-storage-token': 'FE011C19-CF86-4F87-BE5D-9229145D7A06',
- 'x-storage-url': 'https://storage4.clouddrive.com/v1/MossoCloudFS_FE011C19-CF86-4F87-BE5D-9229145D7A06'}
- return (httplib.NO_CONTENT, "", headers, httplib.responses[httplib.NO_CONTENT])
- def _v1_0_slug_servers_72258(self, method, url, body, headers):
- if method != "DELETE":
- raise NotImplemented
- # only used by destroy node()
- return (httplib.ACCEPTED,
- "202 Accepted\n\nThe request is accepted for processing.\n\n ",
- {'date': 'Thu, 09 Jun 2011 10:51:53 GMT', 'content-length': '58',
- 'content-type': 'text/html; charset=UTF-8'},
- httplib.responses[httplib.ACCEPTED])
- def _v1_0_slug_flavors_detail(self, method, url, body, headers):
- body = """<flavors xmlns="http://docs.rackspacecloud.com/servers/api/v1.0">
- <flavor disk="40" id="3" name="m1.medium" ram="4096"/>
- <flavor disk="20" id="2" name="m1.small" ram="2048"/>
- <flavor disk="80" id="4" name="m1.large" ram="8192"/>
- <flavor disk="0" id="6" name="s1" ram="256"/>
- <flavor disk="0" id="7" name="s1.swap" ram="256"/>
- <flavor disk="0" id="1" name="m1.tiny" ram="512"/>
- <flavor disk="10" id="8" name="s1.tiny" ram="512"/>
- <flavor disk="160" id="5" name="m1.xlarge" ram="16384"/>
- </flavors>
- """
- return (httplib.OK, body,
- {'date': 'Tue, 14 Jun 2011 09:43:55 GMT', 'content-length': '529', 'content-type': 'application/xml'},
- httplib.responses[httplib.OK])
+class RackspaceTests(OpenStackTests):
+ should_list_locations = True
+ should_have_pricing = True
+
+ driver_type = RackspaceNodeDriver
+ driver_args = (
+ RACKSPACE_USER,
+ RACKSPACE_KEY,
+ )
if __name__ == '__main__':
Modified: libcloud/trunk/test/test_pricing.py
URL: http://svn.apache.org/viewvc/libcloud/trunk/test/test_pricing.py?rev=1166136&r1=1166135&r2=1166136&view=diff
==============================================================================
--- libcloud/trunk/test/test_pricing.py (original)
+++ libcloud/trunk/test/test_pricing.py Wed Sep 7 11:58:16 2011
@@ -13,11 +13,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import os.path
import sys
import unittest
import libcloud.pricing
+PRICING_FILE_PATH = os.path.join(os.path.dirname(__file__), 'pricing_test.json')
+
class PricingTestCase(unittest.TestCase):
def test_get_pricing_success(self):
@@ -25,7 +28,7 @@ class PricingTestCase(unittest.TestCase)
pricing = libcloud.pricing.get_pricing(driver_type='compute',
driver_name='foo',
- pricing_file_path='test/pricing_test.json')
+ pricing_file_path=PRICING_FILE_PATH)
self.assertEqual(pricing['1'], 1.0)
self.assertEqual(pricing['2'], 2.0)
@@ -55,7 +58,7 @@ class PricingTestCase(unittest.TestCase)
def test_get_pricing_not_in_cache(self):
try:
libcloud.pricing.get_pricing(driver_type='compute', driver_name='inexistent',
- pricing_file_path='test/pricing_test.json')
+ pricing_file_path=PRICING_FILE_PATH)
except KeyError:
pass
else: