| # 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.types import InvalidCredsException |
| from libcloud.drivers.rackspace import RackspaceNodeDriver as Rackspace |
| from libcloud.base import Node, NodeImage, NodeSize |
| |
| from test import MockHttp, TestCaseMixin |
| from test.file_fixtures import FileFixtures |
| |
| from secrets import RACKSPACE_USER, RACKSPACE_KEY |
| import httplib |
| |
| 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 InvalidCredsException, e: |
| self.assertEqual(True, isinstance(e, InvalidCredsException)) |
| 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') |
| |
| def test_list_images(self): |
| ret = self.driver.list_images() |
| self.assertEqual(ret[10].extra['serverId'], None) |
| self.assertEqual(ret[11].extra['serverId'], '91221') |
| |
| 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_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) |
| |
| class RackspaceMockHttp(MockHttp): |
| |
| fixtures = FileFixtures('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_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_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_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_servers_72258(self, method, url, body, headers): |
| if method != "DELETE": |
| raise NotImplemented |
| # only used by destroy node() |
| return (httplib.ACCEPTED, "", {}, httplib.responses[httplib.ACCEPTED]) |
| |
| |
| if __name__ == '__main__': |
| sys.exit(unittest.main()) |
