/opt/imh-python/lib/python3.9/site-packages/libcloud/test/compute
Edit: /opt/imh-python/lib/python3.9/site-packages/libcloud/test/compute/test_gogrid.py (11435B)
# 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.utils.py3 import httplib
from libcloud.utils.py3 import urlparse
from libcloud.utils.py3 import parse_qs
from libcloud.compute.base import NodeState, NodeLocation
from libcloud.common.types import LibcloudError, InvalidCredsError
from libcloud.common.gogrid import GoGridIpAddress
from libcloud.compute.drivers.gogrid import GoGridNodeDriver
from libcloud.compute.base import Node, NodeImage, NodeSize
from libcloud.test import MockHttp # pylint: disable-msg=E0611
from libcloud.test.compute import TestCaseMixin # pylint: disable-msg=E0611
from libcloud.test.file_fixtures import ComputeFileFixtures # pylint: disable-msg=E0611
class GoGridTests(unittest.TestCase, TestCaseMixin):
def setUp(self):
GoGridNodeDriver.connectionCls.conn_class = GoGridMockHttp
GoGridMockHttp.type = None
self.driver = GoGridNodeDriver("foo", "bar")
def _get_test_512Mb_node_size(self):
return NodeSize(id='512Mb',
name=None,
ram=None,
disk=None,
bandwidth=None,
price=None,
driver=self.driver)
def test_create_node(self):
image = NodeImage(1531, None, self.driver)
node = self.driver.create_node(
name='test1',
image=image,
size=self._get_test_512Mb_node_size())
self.assertEqual(node.name, 'test1')
self.assertTrue(node.id is not None)
self.assertEqual(node.extra['password'], 'bebebe')
def test_list_nodes(self):
node = self.driver.list_nodes()[0]
self.assertEqual(node.id, '90967')
self.assertEqual(node.extra['password'], 'bebebe')
self.assertEqual(node.extra['description'], 'test server')
def test_reboot_node(self):
node = Node(90967, None, None, None, None, self.driver)
ret = self.driver.reboot_node(node)
self.assertTrue(ret)
def test_reboot_node_not_successful(self):
GoGridMockHttp.type = 'FAIL'
node = Node(90967, None, None, None, None, self.driver)
try:
self.driver.reboot_node(node)
except Exception:
pass
else:
self.fail('Exception was not thrown')
def test_destroy_node(self):
node = Node(90967, None, None, None, None, self.driver)
ret = self.driver.destroy_node(node)
self.assertTrue(ret)
def test_list_images(self):
images = self.driver.list_images()
image = images[0]
self.assertEqual(len(images), 4)
self.assertEqual(image.name, 'CentOS 5.3 (32-bit) w/ None')
self.assertEqual(image.id, '1531')
location = NodeLocation(
id='gogrid/GSI-939ef909-84b8-4a2f-ad56-02ccd7da05ff.img',
name='test location', country='Slovenia',
driver=self.driver)
images = self.driver.list_images(location=location)
image = images[0]
self.assertEqual(len(images), 4)
self.assertEqual(image.name, 'CentOS 5.3 (32-bit) w/ None')
self.assertEqual(image.id, '1531')
def test_malformed_reply(self):
GoGridMockHttp.type = 'FAIL'
try:
self.driver.list_images()
except LibcloudError as e:
self.assertTrue(isinstance(e, LibcloudError))
else:
self.fail("test should have thrown")
def test_invalid_creds(self):
GoGridMockHttp.type = 'FAIL'
try:
self.driver.list_nodes()
except InvalidCredsError as e:
self.assertTrue(e.driver is not None)
self.assertEqual(e.driver.name, self.driver.name)
else:
self.fail("test should have thrown")
def test_node_creation_without_free_public_ips(self):
GoGridMockHttp.type = 'NOPUBIPS'
try:
image = NodeImage(1531, None, self.driver)
self.driver.create_node(
name='test1',
image=image,
size=self._get_test_512Mb_node_size())
except LibcloudError as e:
self.assertTrue(isinstance(e, LibcloudError))
self.assertTrue(e.driver is not None)
self.assertEqual(e.driver.name, self.driver.name)
else:
self.fail("test should have thrown")
def test_list_locations(self):
locations = self.driver.list_locations()
location_names = [location.name for location in locations]
self.assertEqual(len(locations), 2)
for i in 0, 1:
self.assertTrue(isinstance(locations[i], NodeLocation))
self.assertTrue("US-West-1" in location_names)
self.assertTrue("US-East-1" in location_names)
def test_ex_save_image(self):
node = self.driver.list_nodes()[0]
image = self.driver.ex_save_image(node, "testimage")
self.assertEqual(image.name, "testimage")
def test_ex_edit_image(self):
image = self.driver.list_images()[0]
ret = self.driver.ex_edit_image(image=image, public=False,
ex_description="test", name="testname")
self.assertTrue(isinstance(ret, NodeImage))
def test_ex_edit_node(self):
node = Node(id=90967, name=None, state=None,
public_ips=None, private_ips=None, driver=self.driver)
ret = self.driver.ex_edit_node(node=node,
size=self._get_test_512Mb_node_size())
self.assertTrue(isinstance(ret, Node))
def test_ex_list_ips(self):
ips = self.driver.ex_list_ips()
expected_ips = {"192.168.75.66": GoGridIpAddress(id="5348099",
ip="192.168.75.66", public=True, state="Unassigned",
subnet="192.168.75.64/255.255.255.240"),
"192.168.75.67": GoGridIpAddress(id="5348100",
ip="192.168.75.67", public=True, state="Assigned",
subnet="192.168.75.64/255.255.255.240"),
"192.168.75.68": GoGridIpAddress(id="5348101",
ip="192.168.75.68", public=False, state="Unassigned",
subnet="192.168.75.64/255.255.255.240")}
self.assertEqual(len(expected_ips), 3)
for ip in ips:
self.assertTrue(ip.ip in expected_ips)
self.assertEqual(ip.public, expected_ips[ip.ip].public)
self.assertEqual(ip.state, expected_ips[ip.ip].state)
self.assertEqual(ip.subnet, expected_ips[ip.ip].subnet)
del expected_ips[ip.ip]
self.assertEqual(len(expected_ips), 0)
def test_get_state_invalid(self):
state = self.driver._get_state('invalid')
self.assertEqual(state, NodeState.UNKNOWN)
class GoGridMockHttp(MockHttp):
fixtures = ComputeFileFixtures('gogrid')
def _api_grid_image_list(self, method, url, body, headers):
body = self.fixtures.load('image_list.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_grid_image_list_FAIL(self, method, url, body, headers):
body = "
some non valid json here
"
return (httplib.SERVICE_UNAVAILABLE, body, {},
httplib.responses[httplib.SERVICE_UNAVAILABLE])
def _api_grid_server_list(self, method, url, body, headers):
body = self.fixtures.load('server_list.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
_api_grid_server_list_NOPUBIPS = _api_grid_server_list
def _api_grid_server_list_FAIL(self, method, url, body, headers):
return (httplib.FORBIDDEN,
"123", {}, httplib.responses[httplib.FORBIDDEN])
def _api_grid_ip_list(self, method, url, body, headers):
body = self.fixtures.load('ip_list.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_grid_ip_list_NOPUBIPS(self, method, url, body, headers):
body = self.fixtures.load('ip_list_empty.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_grid_server_power(self, method, url, body, headers):
body = self.fixtures.load('server_power.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_grid_server_power_FAIL(self, method, url, body, headers):
body = self.fixtures.load('server_power_fail.json')
return (httplib.NOT_FOUND, body, {}, httplib.responses[httplib.OK])
def _api_grid_server_add(self, method, url, body, headers):
body = self.fixtures.load('server_add.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
_api_grid_server_add_NOPUBIPS = _api_grid_server_add
def _api_grid_server_delete(self, method, url, body, headers):
body = self.fixtures.load('server_delete.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_grid_server_edit(self, method, url, body, headers):
body = self.fixtures.load('server_edit.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_support_password_list(self, method, url, body, headers):
body = self.fixtures.load('password_list.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
_api_support_password_list_NOPUBIPS = _api_support_password_list
def _api_grid_image_save(self, method, url, body, headers):
body = self.fixtures.load('image_save.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_grid_image_edit(self, method, url, body, headers):
# edit method is quite similar to save method from the response
# perspective
body = self.fixtures.load('image_save.json')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_common_lookup_list(self, method, url, body, headers):
_valid_lookups = ("ip.datacenter",)
lookup = parse_qs(urlparse.urlparse(url).query)["lookup"][0]
if lookup in _valid_lookups:
fixture_path = "lookup_list_%s.json" % \
(lookup.replace(".", "_"))
else:
raise NotImplementedError
body = self.fixtures.load(fixture_path)
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
if __name__ == '__main__':
sys.exit(unittest.main())