| # 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 |
| |
| from libcloud.test import MockHttp, unittest |
| from libcloud.utils.py3 import httplib, assertRaisesRegex |
| from libcloud.common.types import LibcloudError |
| from libcloud.compute.base import NodeState |
| from libcloud.test.secrets import JOYENT_PARAMS |
| from libcloud.test.file_fixtures import ComputeFileFixtures |
| from libcloud.compute.drivers.joyent import JoyentNodeDriver |
| |
| |
| class JoyentTestCase(unittest.TestCase): |
| def setUp(self): |
| JoyentNodeDriver.connectionCls.conn_class = JoyentHttp |
| self.driver = JoyentNodeDriver(*JOYENT_PARAMS) |
| |
| def test_instantiate_multiple_drivers_with_different_region(self): |
| kwargs1 = {"region": "us-east-1"} |
| kwargs2 = {"region": "us-west-1"} |
| driver1 = JoyentNodeDriver(*JOYENT_PARAMS, **kwargs1) |
| driver2 = JoyentNodeDriver(*JOYENT_PARAMS, **kwargs2) |
| |
| self.assertTrue(driver1.connection.host.startswith(kwargs1["region"])) |
| self.assertTrue(driver2.connection.host.startswith(kwargs2["region"])) |
| |
| driver1.list_nodes() |
| driver2.list_nodes() |
| driver1.list_nodes() |
| |
| self.assertTrue(driver1.connection.host.startswith(kwargs1["region"])) |
| self.assertTrue(driver2.connection.host.startswith(kwargs2["region"])) |
| |
| def test_location_backward_compatibility(self): |
| kwargs = {"location": "us-west-1"} |
| driver = JoyentNodeDriver(*JOYENT_PARAMS, **kwargs) |
| self.assertTrue(driver.connection.host.startswith(kwargs["location"])) |
| |
| def test_instantiate_invalid_region(self): |
| expected_msg = "Invalid region.+" |
| |
| assertRaisesRegex( |
| self, |
| LibcloudError, |
| expected_msg, |
| JoyentNodeDriver, |
| "user", |
| "key", |
| location="invalid", |
| ) |
| assertRaisesRegex( |
| self, |
| LibcloudError, |
| expected_msg, |
| JoyentNodeDriver, |
| "user", |
| "key", |
| region="invalid", |
| ) |
| |
| def test_list_sizes(self): |
| sizes = self.driver.list_sizes() |
| |
| self.assertEqual(sizes[0].ram, 16384) |
| |
| def test_list_images(self): |
| images = self.driver.list_images() |
| |
| self.assertEqual(images[0].name, "nodejs") |
| |
| def test_list_nodes_with_and_without_credentials(self): |
| nodes = self.driver.list_nodes() |
| self.assertEqual(len(nodes), 2) |
| |
| node = nodes[0] |
| self.assertEqual(node.public_ips[0], "165.225.129.129") |
| self.assertEqual(node.private_ips[0], "10.112.1.130") |
| self.assertEqual(node.state, NodeState.RUNNING) |
| |
| node = nodes[1] |
| self.assertEqual(node.public_ips[0], "165.225.129.128") |
| self.assertEqual(node.private_ips[0], "10.112.1.131") |
| self.assertEqual(node.state, NodeState.RUNNING) |
| self.assertEqual(node.extra["password"], "abc") |
| |
| def test_create_node(self): |
| image = self.driver.list_images()[0] |
| size = self.driver.list_sizes()[0] |
| node = self.driver.create_node(image=image, size=size, name="testlc") |
| |
| self.assertEqual(node.name, "testlc") |
| |
| def test_ex_stop_node(self): |
| node = self.driver.list_nodes()[0] |
| self.assertTrue(self.driver.ex_stop_node(node)) |
| |
| def test_ex_start_node(self): |
| node = self.driver.list_nodes()[0] |
| self.assertTrue(self.driver.ex_start_node(node)) |
| |
| def test_ex_get_node(self): |
| node_id = "2fb67f5f-53f2-40ab-9d99-b9ff68cfb2ab" |
| node = self.driver.ex_get_node(node_id) |
| self.assertEqual(node.name, "testlc") |
| |
| missing_node = "dummy-node" |
| self.assertRaises(Exception, self.driver.ex_get_node, missing_node, "all") |
| |
| |
| class JoyentHttp(MockHttp): |
| fixtures = ComputeFileFixtures("joyent") |
| |
| def _my_packages(self, method, url, body, headers): |
| body = self.fixtures.load("my_packages.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _my_datasets(self, method, url, body, headers): |
| body = self.fixtures.load("my_datasets.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _my_machines(self, method, url, body, headers): |
| if method == "GET": |
| body = self.fixtures.load("my_machines.json") |
| elif method == "POST": |
| body = self.fixtures.load("my_machines_create.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _my_machines_2fb67f5f_53f2_40ab_9d99_b9ff68cfb2ab(self, method, url, body, headers): |
| body = self.fixtures.load("my_machines_create.json") |
| return (httplib.ACCEPTED, body, {}, httplib.responses[httplib.ACCEPTED]) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |