Add tests for python-cinderclient

Add tests for cinder volume creation and deletion
by ID and by Name.
Add test for cinder show volume output.
Add method to wait for given volume status.
Add method to check that volume deleted.
Add method to check that volume is not deleted (for negative cases).
Add method for volume deletion.
Add method to get Volume properties from CLI output.

Change-Id: I1e1f9bea2afc63e408a88e64fc1b6b6ee9f4e79b
Implements: partial blueprint cinderclient-functional-tests
Co-Authored-by: Kyrylo Romanenko <kromanenko@mirantis.com>

2 files changed, 144 insertions, 3 deletions

diff --git a/cinderclient/tests/functional/base.py b/cinderclient/tests/functional/base.py
index b8d5939..e2b1b67 100644
--- a/cinderclient/tests/functional/base.py
+++ b/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
diff --git a/cinderclient/tests/functional/test_cli.py b/cinderclient/tests/functional/test_cli.py
new file mode 100644
index 0000000..b4beae5
--- /dev/null
+++ b/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'])