Merge "Add tests for python-cinderclient"

cinderclient/tests/functional/base.py

@@ -11,11 +11,12 @@
 #    under the License.
 
 import os
+import time
 
-from six.moves import configparser
+import six
 from tempest_lib.cli import base
 from tempest_lib.cli import output_parser
-
+from tempest_lib import exceptions
 
 _CREDS_FILE = 'functional_creds.conf'
 
@@ -36,7 +37,7 @@ def credentials():
     tenant_name = os.environ.get('OS_TENANT_NAME')
     auth_url = os.environ.get('OS_AUTH_URL')
 
-    config = configparser.RawConfigParser()
+    config = six.moves.configparser.RawConfigParser()
     if config.read(_CREDS_FILE):
         username = username or config.get('admin', 'user')
         password = password or config.get('admin', 'pass')
@@ -95,3 +96,86 @@ class ClientTestBase(base.ClientTestBase):
         for item in items:
             for field in field_names:
                 self.assertIn(field, item)
+
+    def assert_volume_details_rows(self, items):
+        """Check presence of common volume properties.
+
+        :param items: volume properties
+        """
+        values = ('attachments', 'availability_zone', 'bootable', 'created_at',
+                  'description', 'encrypted', 'id', 'metadata', 'name', 'size',
+                  'status', 'user_id', 'volume_type')
+
+        for value in values:
+            self.assertIn(value, items)
+
+    def wait_for_volume_status(self, volume_id, status, timeout=60):
+        """Wait until volume reaches given status.
+
+        :param volume_id: uuid4 id of given volume
+        :param status: expected status of volume
+        :param timeout: timeout in seconds
+        """
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            if status in self.cinder('show', params=volume_id):
+                break
+        else:
+            self.fail("Volume %s did not reach status %s after %d seconds."
+                      % (volume_id, status, timeout))
+
+    def check_volume_not_deleted(self, volume_id):
+        """Check that volume exists.
+
+        :param volume_id: uuid4 id of given volume
+        """
+        self.assertTrue(self.cinder('show', params=volume_id))
+
+    def check_volume_deleted(self, volume_id, timeout=60):
+        """Check that volume deleted successfully.
+
+        :param timeout:
+        :param volume_id: uuid4 id of given volume
+        :param timeout: timeout in seconds
+        """
+        try:
+            start_time = time.time()
+            while time.time() - start_time < timeout:
+                if volume_id not in self.cinder('show', params=volume_id):
+                    break
+        except exceptions.CommandFailed:
+            pass
+        else:
+            self.fail("Volume %s not deleted after %d seconds."
+                      % (volume_id, timeout))
+
+    def volume_create(self, params):
+        """Create volume.
+
+        :param params: parameters to cinder command
+        :return: volume dictionary
+        """
+        output = self.cinder('create', params=params)
+        volume = self._get_property_from_output(output)
+        self.addCleanup(self.volume_delete, volume['id'])
+        self.wait_for_volume_status(volume['id'], 'available')
+        return volume
+
+    def volume_delete(self, volume_id):
+        """Delete specified volume by ID.
+
+        :param volume_id: uuid4 id of given volume
+        """
+        if volume_id in self.cinder('list'):
+            self.cinder('delete', params=volume_id)
+
+    def _get_property_from_output(self, output):
+        """Create a dictionary from an output
+
+        :param output: the output of the cmd
+        """
+        obj = {}
+        items = self.parser.listing(output)
+        for item in items:
+            obj[item['Property']] = six.text_type(item['Value'])
+        return obj

cinderclient/tests/functional/test_cli.py

@@ -0,0 +1,57 @@
+#    Licensed 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.
+
+
+from cinderclient.tests.functional import base
+
+
+class CinderClientTests(base.ClientTestBase):
+    """Basic test for cinder client.
+
+    Check of base cinder commands.
+    """
+    def test_volume_create_delete_id(self):
+        """Create and delete a volume by ID."""
+        volume = self.volume_create(params='1')
+        self.assert_volume_details_rows(volume.keys())
+        self.volume_delete(volume['id'])
+        self.check_volume_deleted(volume['id'])
+
+    def test_volume_create_delete_name(self):
+        """Create and delete a volume by name."""
+        volume = self.volume_create(params='1 --name TestVolumeNamedCreate')
+
+        self.cinder('delete', params='TestVolumeNamedCreate')
+        self.check_volume_deleted(volume['id'])
+
+    def test_volume_show(self):
+        """Show volume details."""
+        volume = self.volume_create(params='1 --name TestVolumeShow')
+        output = self.cinder('show', params='TestVolumeShow')
+        volume = self._get_property_from_output(output)
+        self.assertEqual('TestVolumeShow', volume['name'])
+        self.assert_volume_details_rows(volume.keys())
+
+        self.volume_delete(volume['id'])
+        self.check_volume_deleted(volume['id'])
+
+    def test_volume_extend(self):
+        """Extend a volume size."""
+        volume = self.volume_create(params='1 --name TestVolumeExtend')
+        self.cinder('extend', params="%s %s" % (volume['id'], 2))
+        self.wait_for_volume_status(volume['id'], 'available')
+        output = self.cinder('show', params=volume['id'])
+        volume = self._get_property_from_output(output)
+        self.assertEqual('2', volume['size'])
+
+        self.volume_delete(volume['id'])
+        self.check_volume_deleted(volume['id'])