Commit 03493855 authored by Stavros Sachtouris's avatar Stavros Sachtouris
Browse files

Unittest clients.image create_server

parent db113224
......@@ -92,7 +92,7 @@ class ComputeClientApi(Client):
:returns: request response
"""
data = json_data
if json_data is not None:
if json_data:
data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
self.set_header('Content-Length', len(data))
......
......@@ -30,28 +30,150 @@
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
from mock import patch
from unittest import TestCase
from kamaki.clients import ClientError
import time
from kamaki.clients import livetest, ClientError
from kamaki.clients.cyclades import CycladesClient
class Cyclades(TestCase):
"""Set up a Cyclades thorough test"""
class Cyclades(livetest.Generic):
"""Set up a Cyclades test"""
def setUp(self):
self.url = 'http://cyclades.example.com'
self.token = 'cyc14d3s70k3n'
from kamaki.clients.cyclades import CycladesClient
self.cyclades = CycladesClient(self.url, self.token)
print
with open(self['image', 'details']) as f:
self.img_details = eval(f.read())
self.img = self.img_details['id']
with open(self['flavor', 'details']) as f:
self._flavor_details = eval(f.read())
self.PROFILES = ('ENABLED', 'DISABLED', 'PROTECTED')
self.servers = {}
self.now = time.mktime(time.gmtime())
self.servname1 = 'serv' + unicode(self.now)
self.servname2 = self.servname1 + '_v2'
self.servname1 += '_v1'
self.flavorid = 1
#servers have to be created at the begining...
self.networks = {}
self.netname1 = 'net' + unicode(self.now)
self.netname2 = 'net' + unicode(self.now) + '_v2'
self.client = CycladesClient(self['compute', 'url'], self['token'])
def tearDown(self):
pass
"""Destoy servers used in testing"""
for net in self.networks.keys():
self._delete_network(net)
for server in self.servers.values():
self._delete_server(server['id'])
print('DEL VM %s (%s)' % (server['id'], server['name']))
def test_000(self):
"Prepare a full Cyclades test scenario"
self.server1 = self._create_server(
self.servname1,
self.flavorid,
self.img)
self.server2 = self._create_server(
self.servname2,
self.flavorid + 2,
self.img)
super(self.__class__, self).test_000()
def _create_server(self, servername, flavorid, imageid, personality=None):
server = self.client.create_server(
servername,
flavorid,
imageid,
personality)
print('CREATE VM %s (%s)' % (server['id'], server['name']))
self.servers[servername] = server
return server
def _delete_server(self, servid):
try:
current_state = self.client.get_server_details(servid)
current_state = current_state['status']
if current_state == 'DELETED':
return
self.client.delete_server(servid)
self._wait_for_status(servid, current_state)
self.client.delete_server(servid)
except:
return
def _create_network(self, netname, **kwargs):
net = self.client.create_network(netname, **kwargs)
self.networks[net['id']] = net
return net
def _delete_network(self, netid):
if not netid in self.networks:
return None
print('Disconnect nics of network %s' % netid)
self.client.disconnect_network_nics(netid)
def netwait(wait):
try:
self.client.delete_network(netid)
except ClientError:
time.sleep(wait)
self.do_with_progress_bar(
netwait,
'Delete network %s' % netid,
self._waits[:7])
return self.networks.pop(netid)
def _wait_for_network(self, netid, status):
def netwait(wait):
r = self.client.get_network_details(netid)
if r['status'] == status:
return
time.sleep(wait)
self.do_with_progress_bar(
netwait,
'Wait network %s to reach status %s' % (netid, status),
self._waits[:5])
def _wait_for_nic(self, netid, servid, in_creation=True):
self._wait_for_network(netid, 'ACTIVE')
def nicwait(wait):
nics = self.client.list_server_nics(servid)
for net in nics:
found_nic = net['network_id'] == netid
if (in_creation and found_nic) or not (
in_creation or found_nic):
return
time.sleep(wait)
self.do_with_progress_bar(
nicwait,
'Wait nic-%s-%s to %sconnect' % (
netid,
servid,
'' if in_creation else 'dis'),
self._waits[:5])
for net in self.client.list_server_nics(servid):
if netid == net['network_id']:
return True
return False
def _has_status(self, servid, status):
r = self.client.get_server_details(servid)
return r['status'] == status
def _wait_for_status(self, servid, status):
(wait_bar, wait_cb) = self._safe_progress_bar(
'Server %s in %s' % (servid, status))
self.client.wait_server(servid, status, wait_cb=wait_cb)
self._safe_progress_bar_finish(wait_bar)
"""
def test_parallel_creation(self):
""test create with multiple threads
"""test create with multiple threads
Do not use this in regular livetest
""
"""
from kamaki.clients import SilentEvent
c1 = SilentEvent(
self._create_server,
......@@ -103,7 +225,7 @@ class Cyclades(TestCase):
c8.start()
def test_create_server(self):
""Test create_server""
"""Test create_server"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -118,7 +240,7 @@ class Cyclades(TestCase):
self.assertEqual(self.server1["status"], "BUILD")
def test_list_servers(self):
""Test list servers""
"""Test list servers"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -133,7 +255,7 @@ class Cyclades(TestCase):
servers = self.client.list_servers()
dservers = self.client.list_servers(detail=True)
""detailed and simple are same size""
"""detailed and simple are same size"""
self.assertEqual(len(dservers), len(servers))
for i in range(len(servers)):
for field in (
......@@ -147,19 +269,19 @@ class Cyclades(TestCase):
self.assertFalse(field in servers[i])
self.assertTrue(field in dservers[i])
""detailed and simple contain same names""
"""detailed and simple contain same names"""
names = sorted(map(lambda x: x["name"], servers))
dnames = sorted(map(lambda x: x["name"], dservers))
self.assertEqual(names, dnames)
def _test_0030_wait_test_servers_to_build(self):
""Pseudo-test to wait for VMs to load""
"""Pseudo-test to wait for VMs to load"""
print('')
self._wait_for_status(self.server1['id'], 'BUILD')
self._wait_for_status(self.server2['id'], 'BUILD')
def test_get_server_details(self):
""Test get_server_details""
"""Test get_server_details"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -175,7 +297,7 @@ class Cyclades(TestCase):
self.assertEqual(r["status"], "ACTIVE")
def test_update_server_name(self):
""Test update_server_name""
"""Test update_server_name"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -194,7 +316,7 @@ class Cyclades(TestCase):
self.servers[new_name] = changed
def test_reboot_server(self):
""Test reboot server""
"""Test reboot server"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -216,13 +338,13 @@ class Cyclades(TestCase):
self.assertTrue(self._has_status(self.server2['id'], 'REBOOT'))
def _test_0070_wait_test_servers_to_reboot(self):
""Pseudo-test to wait for VMs to load""
"""Pseudo-test to wait for VMs to load"""
print('')
self._wait_for_status(self.server1['id'], 'REBOOT')
self._wait_for_status(self.server2['id'], 'REBOOT')
def test_create_server_metadata(self):
""Test create_server_metadata""
"""Test create_server_metadata"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -239,7 +361,7 @@ class Cyclades(TestCase):
self.assert_dicts_are_deeply_equal(r1, r2)
def test_get_server_metadata(self):
""Test get server_metadata""
"""Test get server_metadata"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -255,7 +377,7 @@ class Cyclades(TestCase):
self.assertEqual(r['mymeta_0'], 'val_0')
def test_update_server_metadata(self):
""Test update_server_metadata""
"""Test update_server_metadata"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -274,7 +396,7 @@ class Cyclades(TestCase):
self.assertTrue(r2['mymeta3'], 'val3')
def test_delete_server_metadata(self):
""Test delete_server_metadata""
"""Test delete_server_metadata"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -295,7 +417,7 @@ class Cyclades(TestCase):
self.assertEqual(err.status, 404)
def test_list_flavors(self):
""Test flavors_get""
"""Test flavors_get"""
self._test_0120_list_flavors()
def _test_0120_list_flavors(self):
......@@ -305,7 +427,7 @@ class Cyclades(TestCase):
self.assertTrue('SNF:disk_template' in r[0])
def test_get_flavor_details(self):
""Test test_get_flavor_details""
"""Test test_get_flavor_details"""
self._test_0130_get_flavor_details()
def _test_0130_get_flavor_details(self):
......@@ -313,7 +435,7 @@ class Cyclades(TestCase):
self.assert_dicts_are_deeply_equal(self._flavor_details, r)
def test_list_images(self):
""Test list_images""
"""Test list_images"""
self._test_0140_list_images()
def _test_0140_list_images(self):
......@@ -326,7 +448,7 @@ class Cyclades(TestCase):
self.assert_dicts_are_deeply_equal(detailed_img, self.img_details)
def test_get_image_details(self):
""Test image_details""
"""Test image_details"""
self._test_0150_get_image_details()
def _test_0150_get_image_details(self):
......@@ -334,7 +456,7 @@ class Cyclades(TestCase):
self.assert_dicts_are_deeply_equal(r, self.img_details)
def test_get_image_metadata(self):
""Test get_image_metadata""
"""Test get_image_metadata"""
self._test_0160_get_image_metadata()
def _test_0160_get_image_metadata(self):
......@@ -346,7 +468,7 @@ class Cyclades(TestCase):
self.assertEqual(r[key], val)
def test_shutdown_server(self):
""Test shutdown_server""
"""Test shutdown_server"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -361,7 +483,7 @@ class Cyclades(TestCase):
self.assertEqual(r['status'], 'STOPPED')
def test_start_server(self):
""Test start_server""
"""Test start_server"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -378,7 +500,7 @@ class Cyclades(TestCase):
self.assertEqual(r['status'], 'ACTIVE')
def test_get_server_console(self):
""Test get_server_console""
"""Test get_server_console"""
self.server2 = self._create_server(
self.servname2,
self.flavorid + 2,
......@@ -394,7 +516,7 @@ class Cyclades(TestCase):
self.assertTrue('type' in r)
def test_get_firewall_profile(self):
""Test get_firewall_profile""
"""Test get_firewall_profile"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -407,7 +529,7 @@ class Cyclades(TestCase):
self.assertTrue(fprofile in self.PROFILES)
def test_set_firewall_profile(self):
""Test set_firewall_profile""
"""Test set_firewall_profile"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -459,7 +581,7 @@ class Cyclades(TestCase):
self.assertTrue(term in r)
def test_create_network(self):
""Test create_network""
"""Test create_network"""
self._test_0230_create_network()
def _test_0230_create_network(self):
......@@ -490,7 +612,7 @@ class Cyclades(TestCase):
self.assertEqual(self.network1[param], val)
def test_connect_server(self):
""Test connect_server""
"""Test connect_server"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -507,7 +629,7 @@ class Cyclades(TestCase):
self.server1['id']))
def test_disconnect_server(self):
""Test disconnect_server""
"""Test disconnect_server"""
self.test_connect_server()
self._test_0250_disconnect_server()
......@@ -523,7 +645,7 @@ class Cyclades(TestCase):
self._wait_for_network(self.network2['id'], 'ACTIVE')
def test_list_server_nics(self):
""Test list_server_nics""
"""Test list_server_nics"""
self.server1 = self._create_server(
self.servname1,
self.flavorid,
......@@ -544,7 +666,7 @@ class Cyclades(TestCase):
self.assertTrue(len(r) > len0)
def test_list_networks(self):
""Test list_network""
"""Test list_network"""
self.network1 = self._create_network(self.netname1)
self._wait_for_network(self.network1['id'], 'ACTIVE')
self._test_0290_list_networks()
......@@ -569,6 +691,7 @@ class Cyclades(TestCase):
self.assertTrue(term in net.keys())
def test_get_network_details(self):
"""Test get_network_details"""
self.network1 = self._create_network(self.netname1)
self._test_0300_get_network_details()
......@@ -604,7 +727,7 @@ class Cyclades(TestCase):
r = self.client.get_network_details(self.network2['id'])
self.assertEqual(r['name'], updated_name)
"" Don't have auth to test this
""" Don't have auth to test this
def test_delete_image(self):
""Test delete_image""
self._test_0330_delete_image()
......
......@@ -35,7 +35,7 @@ from unittest import makeSuite, TestSuite, TextTestRunner, TestCase
from time import sleep
from kamaki.clients.test.astakos import Astakos
#from kamaki.clients.test.cyclades import Cyclades
from kamaki.clients.test.cyclades import Cyclades
from kamaki.clients.test.image import Image
#from kamaki.clients.test.pithos import Pithos
......@@ -64,13 +64,14 @@ class SilentEvent(TestCase):
while self.can_finish < methodid and wait < 4:
sleep(wait)
wait = 2 * wait
self._value = methodid
self.assertTrue(wait < 4)
def setUp(self):
from kamaki.clients import SilentEvent
self.SE = SilentEvent
def test_parallel(self):
def test_threads(self):
threads = []
for i in range(4):
threads.append(self.SE(self.thread_content, i))
......
This diff is collapsed.
......@@ -128,6 +128,14 @@ example_images_detailed = [
class Image(TestCase):
def assert_dicts_are_deeply_equal(self, d1, d2):
for k, v in d1.items():
self.assertTrue(k in d2)
if isinstance(v, dict):
self.assert_dicts_are_deeply_equal(v, d2[k])
else:
self.assertEqual(unicode(v), unicode(d2[k]))
class FR(object):
json = example_images
headers = {}
......@@ -139,14 +147,10 @@ class Image(TestCase):
pass
def setUp(self):
self.imgname = 'img_%s' % self.now
self.url = 'http://image.example.com'
self.token = 'an1m@g370k3n=='
from kamaki.clients.image import ImageClient
self.client = ImageClient(self.url, self.token)
self.cyclades_url = 'http://cyclades.example.com'
from kamaki.clients.cyclades import CycladesClient
self.cyclades = CycladesClient(self.cyclades_url, self.token)
from kamaki.clients.connection.kamakicon import KamakiHTTPConnection
self.C = KamakiHTTPConnection
......@@ -154,14 +158,6 @@ class Image(TestCase):
self.FR.json = example_images
self.FR.status_code = 200
def assert_dicts_are_deeply_equal(self, d1, d2):
for k, v in d1.items():
self.assertTrue(k in d2)
if isinstance(v, dict):
self.assert_dicts_are_deeply_equal(v, d2[k])
else:
self.assertEqual(unicode(v), unicode(d2[k]))
def test_list_public(self):
with patch.object(
self.C,
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment