Commit ca30409e authored by Stavros Sachtouris's avatar Stavros Sachtouris

Adjust kamaki.clients unittesting

parent efbcdc41
......@@ -38,6 +38,8 @@ import datetime
import os
import sys
from progress.bar import IncrementalBar
from kamaki.clients import ClientError
from kamaki.clients.pithos import PithosClient as pithos
from kamaki.clients.cyclades import CycladesClient as cyclades
......@@ -257,7 +259,7 @@ class testCyclades(unittest.TestCase):
self.img = 'b2dffe52-64a4-48c3-8a4c-8214cc3165cf'
self.img_details = {
u'status': u'ACTIVE',
u'updated': u'2012-10-16T09:04:17+00:00',
u'updated': u'2012-11-19T13:52:16+00:00',
u'name': u'Debian Base',
u'created': u'2012-10-16T09:03:12+00:00',
u'progress': 100,
......@@ -271,8 +273,7 @@ class testCyclades(unittest.TestCase):
u'sortorder': u'1',
u'os': u'debian',
u'root_partition': u'1',
u'description': u'Debian 6.0.6 (Squeeze) Base System',
u'partition_table': u'msdos'}
u'description': u'Debian 6.0.6 (Squeeze) Base System'}
}
}
self.flavor_details = {u'name': u'C1R1024D20',
......@@ -415,53 +416,6 @@ class testCyclades(unittest.TestCase):
return None
return foo
@if_not_all
def test_parallel_creation(self):
"""test create with multiple threads"""
from kamaki.clients import SilentEvent
c1 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c2 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c3 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c4 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c5 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c6 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c7 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c8 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
print('START THREADS')
c1.start()
c2.start()
c3.start()
c4.start()
c5.start()
c6.start()
c7.start()
c8.start()
print('KEWL')
def assert_dicts_are_deeply_equal(self, d1, d2):
for k, v in d1.items():
self.assertTrue(k in d2)
......@@ -631,6 +585,51 @@ class testCyclades(unittest.TestCase):
print('...ok')
"""
@if_not_all
def test_parallel_creation(self):
"""test create with multiple threads"""
from kamaki.clients import SilentEvent
c1 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c2 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c3 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c4 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c5 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c6 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c7 = SilentEvent(self._create_server,
self.servname1,
self.flavorid,
self.img)
c8 = SilentEvent(self._create_server,
self.servname2,
self.flavorid + 2,
self.img)
c1.start()
c2.start()
c3.start()
c4.start()
c5.start()
c6.start()
c7.start()
c8.start()
def _wait_for_network(self, netid, status):
wait = 3
limit = 50
......@@ -683,18 +682,16 @@ class testCyclades(unittest.TestCase):
return r['status'] == status
def _wait_for_status(self, servid, status):
wait = 0
c = ['|', '/', '-', '\\']
while self._has_status(servid, status):
if wait:
sys.stdout.write('\tServer %s in %s. Wait %ss '\
% (servid, status, wait))
for i in range(4 * wait):
sys.stdout.write('\b%s' % c[i % 4])
sys.stdout.flush()
time.sleep(0.25)
print('\b ')
wait = (wait + 3) if wait < 60 else 0
wait_bar = IncrementalBar('\tServer[%s] in %s ' % (servid, status))
wait_bar.start()
def progress_gen(n):
for i in wait_bar.iter(range(int(n))):
yield
yield
self.client.wait_server(servid, status, wait_cb=progress_gen)
wait_bar.finish()
@if_not_all
def test_list_servers(self):
......@@ -899,15 +896,16 @@ class testCyclades(unittest.TestCase):
for detailed_img in r:
if detailed_img['id'] == self.img:
break
self.assert_dicts_are_deeply_equal(r[1], self.img_details)
self.assert_dicts_are_deeply_equal(detailed_img, self.img_details)
@if_not_all
def test_image_details(self):
def test_get_image_details(self):
"""Test image_details"""
self._test_get_image_details
self._test_get_image_details()
def _test_get_image_details(self):
r = self.client.get_image_details(self.img)
r.pop('updated')
self.assert_dicts_are_deeply_equal(r, self.img_details)
@if_not_all
......@@ -1119,8 +1117,6 @@ class testCyclades(unittest.TestCase):
def _test_list_server_nics(self):
r = self.client.list_server_nics(self.server1['id'])
len0 = len(r)
self.assertTrue(len0 > 0)
self.assertTrue('1' in [net['network_id'] for net in r])
self.client.connect_server(self.server1['id'], self.network2['id'])
self.assertTrue(self._wait_for_nic(self.network2['id'],
......
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