Name	Size	Mode	Actions
fixtures/	-	0755	rm
__pycache__/	-	0755	rm
test_abiquo.py	20919	0644	editdlrm
test_auroracompute.py	2189	0644	editdlrm
test_azure.py	22213	0644	editdlrm
test_azure_arm.py	30598	0644	editdlrm
test_base.py	4073	0644	editdlrm
test_bluebox.py	4662	0644	editdlrm
test_brightbox.py	15764	0644	editdlrm
test_bsnl.py	1283	0644	editdlrm
test_cloudscale.py	4540	0644	editdlrm
test_cloudsigma_v1_0.py	8298	0644	editdlrm
test_cloudsigma_v2_0.py	28023	0644	editdlrm
test_cloudstack.py	57399	0644	editdlrm
test_cloudwatt.py	1349	0644	editdlrm
test_deployment.py	29057	0644	editdlrm
test_digitalocean_v2.py	24670	0644	editdlrm
test_dimensiondata_v2_3.py	158501	0644	editdlrm
test_dimensiondata_v2_4.py	163733	0644	editdlrm
test_ec2.py	92953	0644	editdlrm
test_ecp.py	5111	0644	editdlrm
test_ecs.py	47770	0644	editdlrm
test_elasticstack.py	11032	0644	editdlrm
test_exoscale.py	1147	0644	editdlrm
test_gandi.py	14657	0644	editdlrm
test_gce.py	190676	0644	editdlrm
test_gig_g8.py	6826	0644	editdlrm
test_gogrid.py	11435	0644	editdlrm
test_gridscale.py	17208	0644	editdlrm
test_gridspot.py	7682	0644	editdlrm
test_hostvirtual.py	8678	0644	editdlrm
test_ikoula.py	1139	0644	editdlrm
test_indosat.py	1311	0644	editdlrm
test_internetsolutions.py	1371	0644	editdlrm
test_joyent.py	5462	0644	editdlrm
test_kamatera.py	10284	0644	editdlrm
test_kili.py	1525	0644	editdlrm
test_ktucloud.py	4554	0644	editdlrm
test_kubevirt.py	8365	0644	editdlrm
test_libvirt_driver.py	3730	0644	editdlrm
test_linode.py	8778	0644	editdlrm
test_maxihost.py	4306	0644	editdlrm
test_medone.py	1295	0644	editdlrm
test_nephoscale.py	7378	0644	editdlrm
test_ntta.py	1323	0644	editdlrm
test_nttcis.py	148380	0644	editdlrm
test_onapp.py	6175	0644	editdlrm
test_oneandone.py	42388	0644	editdlrm
test_opennebula.py	45456	0644	editdlrm
test_openstack.py	125148	0644	editdlrm
test_ovh.py	10616	0644	editdlrm
test_packet.py	21367	0644	editdlrm
test_profitbricks.py	124039	0644	editdlrm
test_rackspace.py	8015	0644	editdlrm
test_rimuhosting.py	4300	0644	editdlrm
test_scaleway.py	13663	0644	editdlrm
test_softlayer.py	12521	0644	editdlrm
test_ssh_client.py	24225	0644	editdlrm
test_types.py	3202	0644	editdlrm
test_upcloud.py	12056	0644	editdlrm
test_vcl.py	4821	0644	editdlrm
test_vcloud.py	53032	0644	editdlrm
test_voxel.py	6129	0644	editdlrm
test_vpsnet.py	8648	0644	editdlrm
test_vultr.py	6580	0644	editdlrm
__init__.py	4077	0644	editdlrm

# 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 unittest import sys import random import string from libcloud.utils.py3 import httplib from libcloud.compute.drivers.gandi import GandiNodeDriver from libcloud.common.gandi import GandiException from libcloud.compute.types import NodeState from libcloud.test.file_fixtures import ComputeFileFixtures from libcloud.test.secrets import GANDI_PARAMS from libcloud.test.common.test_gandi import BaseGandiMockHttp class GandiTests(unittest.TestCase): node_name = 'test2' def setUp(self): GandiNodeDriver.connectionCls.conn_class = GandiMockHttp GandiMockHttp.type = None self.driver = GandiNodeDriver(*GANDI_PARAMS) def test_list_nodes(self): nodes = self.driver.list_nodes() self.assertTrue(len(nodes) > 0) self.assertTrue(len(nodes[0].public_ips) > 1) def test_list_locations(self): loc = list(filter(lambda x: 'france' in x.country.lower(), self.driver.list_locations()))[0] self.assertEqual(loc.country, 'France') def test_list_images(self): loc = list(filter(lambda x: 'france' in x.country.lower(), self.driver.list_locations()))[0] images = self.driver.list_images(loc) self.assertTrue(len(images) > 2) def test_list_sizes(self): sizes = self.driver.list_sizes() self.assertTrue(len(sizes) >= 1) def test_destroy_node_running(self): nodes = self.driver.list_nodes() test_node = list(filter(lambda x: x.state == NodeState.RUNNING, nodes))[0] self.assertTrue(self.driver.destroy_node(test_node)) def test_destroy_node_halted(self): nodes = self.driver.list_nodes() test_node = list(filter(lambda x: x.state == NodeState.TERMINATED, nodes))[0] self.assertTrue(self.driver.destroy_node(test_node)) def test_reboot_node(self): nodes = self.driver.list_nodes() test_node = list(filter(lambda x: x.state == NodeState.RUNNING, nodes))[0] self.assertTrue(self.driver.reboot_node(test_node)) def test_create_node(self): login = 'libcloud' passwd = ''.join(random.choice(string.ascii_letters) for i in range(10)) # Get france datacenter loc = list(filter(lambda x: 'france' in x.country.lower(), self.driver.list_locations()))[0] # Get a debian image images = self.driver.list_images(loc) images = [x for x in images if x.name.lower().startswith('debian')] img = list(filter(lambda x: '5' in x.name, images))[0] # Get a configuration size size = self.driver.list_sizes()[0] node = self.driver.create_node(name=self.node_name, login=login, password=passwd, image=img, location=loc, size=size) self.assertEqual(node.name, self.node_name) def test_create_volume(self): loc = list(filter(lambda x: 'france' in x.country.lower(), self.driver.list_locations()))[0] volume = self.driver.create_volume( size=1024, name='libcloud', location=loc) self.assertEqual(volume.name, 'libcloud') self.assertEqual(volume.size, 1024) def test_list_volumes(self): disks = self.driver.list_volumes() self.assertTrue(len(disks) > 0) def test_destroy_volume(self): volumes = self.driver.list_volumes() test_vol = list(filter(lambda x: x.name == 'test_disk', volumes))[0] self.assertTrue(self.driver.destroy_volume(test_vol)) def test_attach_volume(self): disks = self.driver.list_volumes() nodes = self.driver.list_nodes() res = self.driver.attach_volume(nodes[0], disks[0]) self.assertTrue(res) def test_detach_volume(self): disks = self.driver.list_volumes() nodes = self.driver.list_nodes() res = self.driver.detach_volume(nodes[0], disks[0]) self.assertTrue(res) def test_ex_list_interfaces(self): ifaces = self.driver.ex_list_interfaces() self.assertTrue(len(ifaces) > 0) def test_ex_attach_interface(self): ifaces = self.driver.ex_list_interfaces() nodes = self.driver.list_nodes() res = self.driver.ex_node_attach_interface(nodes[0], ifaces[0]) self.assertTrue(res) def test_ex_detach_interface(self): ifaces = self.driver.ex_list_interfaces() nodes = self.driver.list_nodes() res = self.driver.ex_node_detach_interface(nodes[0], ifaces[0]) self.assertTrue(res) def test_ex_snapshot_disk(self): disks = self.driver.list_volumes() self.assertTrue(self.driver.ex_snapshot_disk(disks[2])) self.assertRaises(GandiException, self.driver.ex_snapshot_disk, disks[0]) def test_ex_update_disk(self): disks = self.driver.list_volumes() self.assertTrue(self.driver.ex_update_disk(disks[0], new_size=4096)) def test_list_key_pairs(self): keys = self.driver.list_key_pairs() self.assertTrue(len(keys) > 0) def test_get_key_pair(self): key = self.driver.get_key_pair(10) self.assertEqual(key.name, 'testkey') def test_import_key_pair_from_string(self): key = self.driver.import_key_pair_from_string('testkey', '12345') self.assertEqual(key.name, 'testkey') self.assertEqual(key.extra['id'], 10) def test_delete_key_pair(self): response = self.driver.delete_key_pair(10) self.assertTrue(response) def test_ex_get_node(self): node = self.driver.ex_get_node(34951) self.assertEqual(node.name, "test2") def test_ex_get_volume(self): volume = self.driver.ex_get_volume(1263) self.assertEqual(volume.name, "libcloud") class GandiRatingTests(unittest.TestCase): """Tests where rating model is involved""" node_name = 'test2' def setUp(self): GandiNodeDriver.connectionCls.conn_class = GandiMockRatingHttp GandiMockRatingHttp.type = None self.driver = GandiNodeDriver(*GANDI_PARAMS) def test_list_sizes(self): sizes = self.driver.list_sizes() self.assertEqual(len(sizes), 4) def test_create_node(self): login = 'libcloud' passwd = ''.join(random.choice(string.ascii_letters) for i in range(10)) # Get france datacenter loc = list(filter(lambda x: 'france' in x.country.lower(), self.driver.list_locations()))[0] # Get a debian image images = self.driver.list_images(loc) images = [x for x in images if x.name.lower().startswith('debian')] img = list(filter(lambda x: '5' in x.name, images))[0] # Get a configuration size size = self.driver.list_sizes()[0] node = self.driver.create_node(name=self.node_name, login=login, password=passwd, image=img, location=loc, size=size) self.assertEqual(node.name, self.node_name) class GandiMockHttp(BaseGandiMockHttp): fixtures = ComputeFileFixtures('gandi') def _xmlrpc__hosting_datacenter_list(self, method, url, body, headers): body = self.fixtures.load('datacenter_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_image_list(self, method, url, body, headers): body = self.fixtures.load('image_list_dc0.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_list(self, method, url, body, headers): body = self.fixtures.load('vm_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_ip_list(self, method, url, body, headers): body = self.fixtures.load('ip_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_account_info(self, method, url, body, headers): body = self.fixtures.load('account_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_info(self, method, url, body, headers): body = self.fixtures.load('vm_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_delete(self, method, url, body, headers): body = self.fixtures.load('vm_delete.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__operation_info(self, method, url, body, headers): body = self.fixtures.load('operation_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_create_from(self, method, url, body, headers): body = self.fixtures.load('vm_create_from.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_reboot(self, method, url, body, headers): body = self.fixtures.load('vm_reboot.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_stop(self, method, url, body, headers): body = self.fixtures.load('vm_stop.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_iface_list(self, method, url, body, headers): body = self.fixtures.load('iface_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_disk_list(self, method, url, body, headers): body = self.fixtures.load('disk_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_iface_attach(self, method, url, body, headers): body = self.fixtures.load('iface_attach.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_iface_detach(self, method, url, body, headers): body = self.fixtures.load('iface_detach.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_disk_attach(self, method, url, body, headers): body = self.fixtures.load('disk_attach.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_disk_detach(self, method, url, body, headers): body = self.fixtures.load('disk_detach.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_disk_create(self, method, url, body, headers): body = self.fixtures.load('disk_create.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_disk_create_from(self, method, url, body, headers): body = self.fixtures.load('disk_create_from.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_disk_info(self, method, url, body, headers): body = self.fixtures.load('disk_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_disk_update(self, method, url, body, headers): body = self.fixtures.load('disk_update.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_disk_delete(self, method, url, body, headers): body = self.fixtures.load('disk_delete.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_ssh_info(self, method, url, body, headers): body = self.fixtures.load('ssh_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_ssh_list(self, method, url, body, headers): body = self.fixtures.load('ssh_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_ssh_create(self, method, url, body, headers): body = self.fixtures.load('ssh_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_ssh_delete(self, method, url, body, headers): body = self.fixtures.load('ssh_delete.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) class GandiMockRatingHttp(BaseGandiMockHttp): """Fixtures needed for tests related to rating model""" fixtures = ComputeFileFixtures('gandi') def _xmlrpc__hosting_datacenter_list(self, method, url, body, headers): body = self.fixtures.load('datacenter_list.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_image_list(self, method, url, body, headers): body = self.fixtures.load('image_list_dc0.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_create_from(self, method, url, body, headers): body = self.fixtures.load('vm_create_from.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__operation_info(self, method, url, body, headers): body = self.fixtures.load('operation_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) def _xmlrpc__hosting_vm_info(self, method, url, body, headers): body = self.fixtures.load('vm_info.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) # Specific to rating tests def _xmlrpc__hosting_account_info(self, method, url, body, headers): body = self.fixtures.load('account_info_rating.xml') return (httplib.OK, body, {}, httplib.responses[httplib.OK]) if __name__ == '__main__': sys.exit(unittest.main())