Merge "Fixes bug 960864- Testcases for the action list Volumes and list Volumes with Detail"
diff --git a/tempest/services/nova/json/volumes_client.py b/tempest/services/nova/json/volumes_client.py
index 95131f2..c935621 100644
--- a/tempest/services/nova/json/volumes_client.py
+++ b/tempest/services/nova/json/volumes_client.py
@@ -1,5 +1,7 @@
+from tempest import exceptions
from tempest.common import rest_client
import json
+import time
class VolumesClient(object):
@@ -9,6 +11,10 @@
catalog_type = self.config.compute.catalog_type
self.client = rest_client.RestClient(config, username, key, auth_url,
catalog_type, tenant_name)
+ self.build_interval = self.config.compute.build_interval
+ self.build_timeout = self.config.compute.build_timeout
+ self.headers = {'Content-Type': 'application/json',
+ 'Accept': 'application/json'}
def list_volumes(self, params=None):
"""List all the volumes created"""
@@ -24,6 +30,20 @@
body = json.loads(body)
return resp, body['volumes']
+ def list_volumes_with_detail(self, params=None):
+ """List all the details of volumes"""
+ url = 'os-volumes/detail'
+ if params != None:
+ param_list = []
+ for param, value in params.iteritems():
+ param_list.append("%s=%s&" % (param, value))
+
+ url = '?' + ' '.join(param_list)
+
+ resp, body = self.client.get(url)
+ body = json.loads(body)
+ return resp, body['volumes']
+
def get_volume(self, volume_id):
"""Returns the details of a single volume"""
url = "os-volumes/%s" % str(volume_id)
@@ -31,6 +51,45 @@
body = json.loads(body)
return resp, body['volume']
+ def create_volume(self, size, **kwargs):
+ """
+ Creates a new Volume.
+ size(Required): Size of volume in GB.
+ Following optional keyword arguments are accepted:
+ display_name: Optional Volume Name.
+ metadata: A dictionary of values to be used as metadata.
+ """
+ post_body = {
+ 'size': size,
+ 'display_name': kwargs.get('display_name'),
+ 'metadata': kwargs.get('metadata'),
+ }
+
+ post_body = json.dumps({'volume': post_body})
+ resp, body = self.client.post('os-volumes', post_body, self.headers)
+ body = json.loads(body)
+ return resp, body['volume']
+
def delete_volume(self, volume_id):
"""Deletes the Specified Volume"""
return self.client.delete("os-volumes/%s" % str(volume_id))
+
+ def wait_for_volume_status(self, volume_id, status):
+ """Waits for a Volume to reach a given status"""
+ resp, body = self.get_volume(volume_id)
+ volume_name = body['displayName']
+ volume_status = body['status']
+ start = int(time.time())
+
+ while volume_status != status:
+ time.sleep(self.build_interval)
+ resp, body = self.get_volume(volume_id)
+ volume_status = body['status']
+ if volume_status == 'error':
+ raise exceptions.BuildErrorException(volume_id=volume_id)
+
+ if int(time.time()) - start >= self.build_timeout:
+ message = 'Volume %s failed to reach %s status within '\
+ 'the required time (%s s).' % (volume_name, status,
+ self.build_timeout)
+ raise exceptions.TimeoutException(message)
diff --git a/tempest/tests/test_volumes_list.py b/tempest/tests/test_volumes_list.py
new file mode 100644
index 0000000..4a4c7d1
--- /dev/null
+++ b/tempest/tests/test_volumes_list.py
@@ -0,0 +1,59 @@
+from nose.plugins.attrib import attr
+import unittest2 as unittest
+from tempest import openstack
+from tempest.common.utils.data_utils import rand_name
+
+
+class VolumesTest(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ cls.os = openstack.Manager()
+ cls.client = cls.os.volumes_client
+ #Create 3 Volumes
+ cls.volume_list = list()
+ cls.volume_id_list = list()
+ for i in range(3):
+ v_name = rand_name('Name-')
+ metadata = {'Type': 'work'}
+ resp, volume = cls.client.create_volume(size=1,
+ display_name=v_name,
+ metadata=metadata)
+ cls.client.wait_for_volume_status(volume['id'],
+ 'available')
+ resp, volume = cls.client.get_volume(volume['id'])
+ cls.volume_list.append(volume)
+ cls.volume_id_list.append(volume['id'])
+
+ def teardown(cls):
+ #Delete the created Volumes
+ for i in range(3):
+ resp, _ = cls.client.delete_volume(cls.volume_id_list[i])
+
+ @attr(type='smoke')
+ def test_volume_list(self):
+ """Should return the list of Volumes"""
+ #Fetch all Volumes
+ resp, fetched_list = self.client.list_volumes()
+ self.assertEqual(200, resp.status)
+ #Now check if all the Volumes created in setup are in fetched list
+ missing_volumes =\
+ [v for v in self.volume_list if v not in fetched_list]
+ self.assertFalse(missing_volumes,
+ "Failed to find volume %s in fetched list"
+ % ', '.join(m_vol['displayName']
+ for m_vol in missing_volumes))
+
+ @attr(type='smoke')
+ def test_volume_list_with_details(self):
+ """Should return the list of Volumes with details"""
+ #Fetch all Volumes
+ resp, fetched_list = self.client.list_volumes_with_detail()
+ self.assertEqual(200, resp.status)
+ #Now check if all the Volumes created in setup are in fetched list
+ missing_volumes =\
+ [v for v in self.volume_list if v not in fetched_list]
+ self.assertFalse(missing_volumes,
+ "Failed to find volume %s in fetched list"
+ % ', '.join(m_vol['displayName']
+ for m_vol in missing_volumes))