| # 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.py3 import httplib |
| |
| from libcloud.common.types import InvalidCredsError |
| from libcloud.compute.drivers.opsource import OpsourceNodeDriver as Opsource |
| from libcloud.compute.drivers.opsource import OpsourceAPIException |
| from libcloud.compute.base import Node, NodeAuthPassword, NodeLocation |
| |
| from test import MockHttp |
| from test.compute import TestCaseMixin |
| from test.file_fixtures import ComputeFileFixtures |
| |
| from test.secrets import OPSOURCE_PARAMS |
| |
| class OpsourceTests(unittest.TestCase, TestCaseMixin): |
| |
| def setUp(self): |
| Opsource.connectionCls.conn_classes = (None, OpsourceMockHttp) |
| OpsourceMockHttp.type = None |
| self.driver = Opsource(*OPSOURCE_PARAMS) |
| |
| def test_invalid_creds(self): |
| OpsourceMockHttp.type = 'UNAUTHORIZED' |
| try: |
| self.driver.list_nodes() |
| self.assertTrue(False) # Above command should have thrown an InvalidCredsException |
| except InvalidCredsError: |
| self.assertTrue(True) |
| |
| def test_list_sizes_response(self): |
| OpsourceMockHttp.type = None |
| ret = self.driver.list_sizes() |
| self.assertEqual(len(ret), 1) |
| size = ret[0] |
| self.assertEqual(size.name, 'default') |
| |
| def test_reboot_node_response(self): |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| ret = node.reboot() |
| self.assertTrue(ret is True) |
| |
| def test_reboot_node_response_INPROGRESS(self): |
| OpsourceMockHttp.type = 'INPROGRESS' |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| try: |
| node.reboot() |
| self.assertTrue(False) # above command should have thrown OpsourceAPIException |
| except OpsourceAPIException: |
| self.assertTrue(True) |
| |
| def test_destroy_node_response(self): |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| ret = node.destroy() |
| self.assertTrue(ret is True) |
| |
| def test_destroy_node_response_INPROGRESS(self): |
| OpsourceMockHttp.type = 'INPROGRESS' |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| try: |
| node.destroy() |
| self.assertTrue(False) # above command should have thrown OpsourceAPIException |
| except OpsourceAPIException: |
| self.assertTrue(True) |
| |
| def test_create_node_response(self): |
| rootPw = NodeAuthPassword('pass123') |
| image = self.driver.list_images()[0] |
| network = self.driver.ex_list_networks()[0] |
| node = self.driver.create_node(name='test2', image=image, auth=rootPw, |
| ex_description='test2 node', ex_network=network, |
| ex_isStarted=False) |
| self.assertEqual(node.id, 'e75ead52-692f-4314-8725-c8a4f4d13a87') |
| self.assertEqual(node.extra['status'].action, 'DEPLOY_SERVER') |
| |
| def test_ex_shutdown_graceful(self): |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| ret = self.driver.ex_shutdown_graceful(node) |
| self.assertTrue(ret is True) |
| |
| def test_ex_shutdown_graceful_INPROGRESS(self): |
| OpsourceMockHttp.type = 'INPROGRESS' |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| try: |
| self.driver.ex_shutdown_graceful(node) |
| self.assertTrue(False) # above command should have thrown OpsourceAPIException |
| except OpsourceAPIException: |
| self.assertTrue(True) |
| |
| def test_ex_start_node(self): |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| ret = self.driver.ex_start_node(node) |
| self.assertTrue(ret is True) |
| |
| def test_ex_start_node_INPROGRESS(self): |
| OpsourceMockHttp.type = 'INPROGRESS' |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| try: |
| self.driver.ex_start_node(node) |
| self.assertTrue(False) # above command should have thrown OpsourceAPIException |
| except OpsourceAPIException: |
| self.assertTrue(True) |
| |
| def test_ex_power_off(self): |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| ret = self.driver.ex_power_off(node) |
| self.assertTrue(ret is True) |
| |
| def test_ex_power_off_INPROGRESS(self): |
| OpsourceMockHttp.type = 'INPROGRESS' |
| node = Node(id='11', name=None, state=None, |
| public_ips=None, private_ips=None, driver=self.driver) |
| try: |
| self.driver.ex_power_off(node) |
| self.assertTrue(False) # above command should have thrown OpsourceAPIException |
| except OpsourceAPIException: |
| self.assertTrue(True) |
| |
| def test_ex_list_networks(self): |
| nets = self.driver.ex_list_networks() |
| self.assertEqual(nets[0].name, 'test-net1') |
| self.assertTrue(isinstance(nets[0].location, NodeLocation)) |
| |
| class OpsourceMockHttp(MockHttp): |
| |
| fixtures = ComputeFileFixtures('opsource') |
| |
| def _oec_0_9_myaccount_UNAUTHORIZED(self, method, url, body, headers): |
| return (httplib.UNAUTHORIZED, "", {}, httplib.responses[httplib.UNAUTHORIZED]) |
| |
| def _oec_0_9_myaccount(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_myaccount.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_myaccount_INPROGRESS(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_myaccount.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_base_image(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_base_image.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_deployed(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_deployed.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_pendingDeploy(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_pendingDeploy.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_datacenter(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_datacenter.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11(self, method, url, body, headers): |
| body = None |
| action = url.split('?')[-1] |
| |
| if action == 'restart': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_restart.xml') |
| elif action == 'shutdown': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_shutdown.xml') |
| elif action == 'delete': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_delete.xml') |
| elif action == 'start': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_start.xml') |
| elif action == 'poweroff': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_poweroff.xml') |
| |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_INPROGRESS(self, method, url, body, headers): |
| body = None |
| action = url.split('?')[-1] |
| |
| if action == 'restart': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_restart_INPROGRESS.xml') |
| elif action == 'shutdown': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_shutdown_INPROGRESS.xml') |
| elif action == 'delete': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_delete_INPROGRESS.xml') |
| elif action == 'start': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_start_INPROGRESS.xml') |
| elif action == 'poweroff': |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_11_poweroff_INPROGRESS.xml') |
| |
| return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server(self, method, url, body, headers): |
| body = self.fixtures.load('_oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_networkWithLocation(self, method, url, body, headers): |
| body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_networkWithLocation.xml') |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| |
| if __name__ == '__main__': |
| sys.exit(unittest.main()) |
