| # 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 json |
| from unittest.mock import Mock, call |
| |
| from libcloud.test import unittest |
| from libcloud.compute.base import NodeSize, NodeImage, NodeLocation, NodeAuthSSHKey |
| from libcloud.common.upcloud import ( |
| PlanPrice, |
| UpcloudNodeDestroyer, |
| UpcloudNodeOperations, |
| UpcloudTimeoutException, |
| UpcloudCreateNodeRequestBody, |
| _StorageDevice, |
| ) |
| |
| |
| class TestUpcloudCreateNodeRequestBody(unittest.TestCase): |
| def setUp(self): |
| self.image = NodeImage( |
| id="01000000-0000-4000-8000-000030060200", |
| name="Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| driver="", |
| extra={"type": "template"}, |
| ) |
| self.location = NodeLocation(id="fi-hel1", name="Helsinki #1", country="FI", driver="") |
| self.size = NodeSize( |
| id="1xCPU-1GB", |
| name="1xCPU-1GB", |
| ram=1024, |
| disk=30, |
| bandwidth=2048, |
| extra={"core_number": 1, "storage_tier": "maxiops"}, |
| price=None, |
| driver="", |
| ) |
| |
| def test_creating_node_from_template_image(self): |
| body = UpcloudCreateNodeRequestBody( |
| name="ts", image=self.image, location=self.location, size=self.size |
| ) |
| json_body = body.to_json() |
| dict_body = json.loads(json_body) |
| expected_body = { |
| "server": { |
| "title": "ts", |
| "hostname": "localhost", |
| "plan": "1xCPU-1GB", |
| "zone": "fi-hel1", |
| "login_user": {"username": "root", "create_password": "yes"}, |
| "storage_devices": { |
| "storage_device": [ |
| { |
| "action": "clone", |
| "title": "Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| "storage": "01000000-0000-4000-8000-000030060200", |
| "size": 30, |
| "tier": "maxiops", |
| } |
| ] |
| }, |
| } |
| } |
| self.assertDictEqual(expected_body, dict_body) |
| |
| def test_creating_node_from_cdrom_image(self): |
| image = NodeImage( |
| id="01000000-0000-4000-8000-000030060200", |
| name="Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| driver="", |
| extra={"type": "cdrom"}, |
| ) |
| body = UpcloudCreateNodeRequestBody( |
| name="ts", image=image, location=self.location, size=self.size |
| ) |
| json_body = body.to_json() |
| dict_body = json.loads(json_body) |
| expected_body = { |
| "server": { |
| "title": "ts", |
| "hostname": "localhost", |
| "plan": "1xCPU-1GB", |
| "zone": "fi-hel1", |
| "login_user": {"username": "root", "create_password": "yes"}, |
| "storage_devices": { |
| "storage_device": [ |
| { |
| "action": "create", |
| "size": 30, |
| "tier": "maxiops", |
| "title": "Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| }, |
| { |
| "action": "attach", |
| "storage": "01000000-0000-4000-8000-000030060200", |
| "type": "cdrom", |
| }, |
| ] |
| }, |
| } |
| } |
| self.assertDictEqual(expected_body, dict_body) |
| |
| def test_creating_node_using_ssh_keys(self): |
| auth = NodeAuthSSHKey("sshkey") |
| |
| body = UpcloudCreateNodeRequestBody( |
| name="ts", |
| image=self.image, |
| location=self.location, |
| size=self.size, |
| auth=auth, |
| ) |
| json_body = body.to_json() |
| dict_body = json.loads(json_body) |
| expected_body = { |
| "server": { |
| "title": "ts", |
| "hostname": "localhost", |
| "plan": "1xCPU-1GB", |
| "zone": "fi-hel1", |
| "login_user": { |
| "username": "root", |
| "ssh_keys": {"ssh_key": ["sshkey"]}, |
| }, |
| "storage_devices": { |
| "storage_device": [ |
| { |
| "action": "clone", |
| "size": 30, |
| "title": "Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| "tier": "maxiops", |
| "storage": "01000000-0000-4000-8000-000030060200", |
| } |
| ] |
| }, |
| } |
| } |
| self.assertDictEqual(expected_body, dict_body) |
| |
| def test_creating_node_using_hostname(self): |
| body = UpcloudCreateNodeRequestBody( |
| name="ts", |
| image=self.image, |
| location=self.location, |
| size=self.size, |
| ex_hostname="myhost.upcloud.com", |
| ) |
| json_body = body.to_json() |
| dict_body = json.loads(json_body) |
| expected_body = { |
| "server": { |
| "title": "ts", |
| "hostname": "myhost.upcloud.com", |
| "plan": "1xCPU-1GB", |
| "zone": "fi-hel1", |
| "login_user": {"username": "root", "create_password": "yes"}, |
| "storage_devices": { |
| "storage_device": [ |
| { |
| "action": "clone", |
| "title": "Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| "storage": "01000000-0000-4000-8000-000030060200", |
| "tier": "maxiops", |
| "size": 30, |
| } |
| ] |
| }, |
| } |
| } |
| self.assertDictEqual(expected_body, dict_body) |
| |
| def test_creating_node_with_non_default_username(self): |
| body = UpcloudCreateNodeRequestBody( |
| name="ts", |
| image=self.image, |
| location=self.location, |
| size=self.size, |
| ex_username="someone", |
| ) |
| json_body = body.to_json() |
| dict_body = json.loads(json_body) |
| |
| login_user = dict_body["server"]["login_user"] |
| self.assertDictEqual({"username": "someone", "create_password": "yes"}, login_user) |
| |
| |
| class TestStorageDevice(unittest.TestCase): |
| def setUp(self): |
| self.image = NodeImage( |
| id="01000000-0000-4000-8000-000030060200", |
| name="Ubuntu Server 16.04 LTS (Xenial Xerus)", |
| driver="", |
| extra={"type": "template"}, |
| ) |
| self.size = NodeSize( |
| id="1xCPU-1GB", |
| name="1xCPU-1GB", |
| ram=1024, |
| disk=30, |
| bandwidth=2048, |
| extra={"core_number": 1}, |
| price=None, |
| driver="", |
| ) |
| |
| def test_storage_tier_default_value(self): |
| storagedevice = _StorageDevice(self.image, self.size) |
| d = storagedevice.to_dict() |
| |
| self.assertEqual(d["storage_device"][0]["tier"], "maxiops") |
| |
| def test_storage_tier_given(self): |
| self.size.extra["storage_tier"] = "hdd" |
| storagedevice = _StorageDevice(self.image, self.size) |
| d = storagedevice.to_dict() |
| |
| self.assertEqual(d["storage_device"][0]["tier"], "hdd") |
| |
| |
| class TestUpcloudNodeDestroyer(unittest.TestCase): |
| def setUp(self): |
| self.mock_sleep = Mock() |
| self.mock_operations = Mock(spec=UpcloudNodeOperations) |
| self.destroyer = UpcloudNodeDestroyer(self.mock_operations, sleep_func=self.mock_sleep) |
| |
| def test_node_already_in_stopped_state(self): |
| self.mock_operations.get_node_state.side_effect = ["stopped"] |
| |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| |
| self.assertTrue(self.mock_operations.stop_node.call_count == 0) |
| self.mock_operations.destroy_node.assert_called_once_with(1) |
| |
| def test_node_in_error_state(self): |
| self.mock_operations.get_node_state.side_effect = ["error"] |
| |
| self.assertFalse(self.destroyer.destroy_node(1)) |
| |
| self.assertTrue(self.mock_operations.stop_node.call_count == 0) |
| self.assertTrue(self.mock_operations.destroy_node.call_count == 0) |
| |
| def test_node_in_started_state(self): |
| self.mock_operations.get_node_state.side_effect = ["started", "stopped"] |
| |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| |
| self.mock_operations.stop_node.assert_called_once_with(1) |
| self.mock_operations.destroy_node.assert_called_once_with(1) |
| |
| def test_node_in_maintenace_state(self): |
| self.mock_operations.get_node_state.side_effect = [ |
| "maintenance", |
| "maintenance", |
| None, |
| ] |
| |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| |
| self.mock_sleep.assert_has_calls( |
| [call(self.destroyer.WAIT_AMOUNT), call(self.destroyer.WAIT_AMOUNT)] |
| ) |
| |
| self.assertTrue(self.mock_operations.stop_node.call_count == 0) |
| self.assertTrue(self.mock_operations.destroy_node.call_count == 0) |
| |
| def test_node_statys_in_started_state_for_awhile(self): |
| self.mock_operations.get_node_state.side_effect = [ |
| "started", |
| "started", |
| "stopped", |
| ] |
| |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| |
| # Only one all for stop should be done |
| self.mock_operations.stop_node.assert_called_once_with(1) |
| self.mock_sleep.assert_has_calls([call(self.destroyer.WAIT_AMOUNT)]) |
| self.mock_operations.destroy_node.assert_called_once_with(1) |
| |
| def test_reuse(self): |
| """Verify that internal flag self.destroyer._stop_node is handled properly""" |
| self.mock_operations.get_node_state.side_effect = [ |
| "started", |
| "stopped", |
| "started", |
| "stopped", |
| ] |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| |
| self.assertEqual(self.mock_sleep.call_count, 0) |
| self.assertEqual(self.mock_operations.stop_node.call_count, 2) |
| |
| def test_timeout(self): |
| self.mock_operations.get_node_state.side_effect = ["maintenance"] * 50 |
| |
| self.assertRaises(UpcloudTimeoutException, self.destroyer.destroy_node, 1) |
| |
| def test_timeout_reuse(self): |
| """Verify sleep count is handled properly""" |
| self.mock_operations.get_node_state.side_effect = ["maintenance"] * 50 |
| self.assertRaises(UpcloudTimeoutException, self.destroyer.destroy_node, 1) |
| |
| self.mock_operations.get_node_state.side_effect = ["maintenance", None] |
| self.assertTrue(self.destroyer.destroy_node(1)) |
| |
| |
| class TestPlanPrice(unittest.TestCase): |
| def setUp(self): |
| prices = [ |
| {"name": "uk-lon1", "server_plan_1xCPU-1GB": {"amount": 1, "price": 1.488}}, |
| {"name": "fi-hel1", "server_plan_1xCPU-1GB": {"amount": 1, "price": 1.588}}, |
| ] |
| self.pp = PlanPrice(prices) |
| |
| def test_zone_prices(self): |
| location = NodeLocation(id="fi-hel1", name="Helsinki #1", country="FI", driver=None) |
| self.assertEqual(self.pp.get_price("1xCPU-1GB", location), 1.588) |
| |
| def test_plan_not_found_in_zone(self): |
| location = NodeLocation(id="no_such_location", name="", country="", driver=None) |
| self.assertIsNone(self.pp.get_price("1xCPU-1GB", location)) |
| |
| def test_no_location_given(self): |
| self.assertIsNone(self.pp.get_price("1xCPU-1GB")) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
