oci-iscsi-config auto tests

Author: guidotijskens

lib/oci_utils/impl/oci-iscsi-config-main.py

@@ -59,10 +59,10 @@ def call(cmd, log_output=True):
     _c = _prepare_command(cmd)
     try:
         if _logger.isEnabledFor(logging.DEBUG):
-            _logger.debug('Executing [%s]' % ' '.join(_c))
+            _logger.debug('Executing [%s]', ' '.join(_c))
         cp = subprocess.run(_c, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
         if cp.returncode != 0 and log_output:
-            _logger.debug("execution failed: ec=%s, output=[%s], stderr=[%s] " % (cp.returncode, cp.stdout, cp.stderr))
+            _logger.debug("execution failed: ec=%s, output=[%s], stderr=[%s] ", cp.returncode, cp.stdout, cp.stderr)
         return cp.returncode
     except OSError:
         return 404
@@ -97,7 +97,7 @@ def call_output(cmd, log_output=True):
         if log_output:
             if _logger.isEnabledFor(logging.DEBUG):
                 # pylint: disable=logging-not-lazy,logging-format-interpolation
-                _logger.debug("Error execeuting {}: {}\n{}\n".format(_c, e.returncode, e.output))
+                _logger.debug("Error executing {}: {}\n{}\n".format(_c, e.returncode, e.output.decode('utf-8')))
         return None
 
 
@@ -207,7 +207,7 @@ def write_to_file(path, content):
                                 stderr=subprocess.PIPE,
                                 stdin=subprocess.PIPE).communicate(content.encode())
     if err:
-        _logger.debug("Error writing content to file: %s" % err)
+        _logger.debug("Error writing content to file: %s", err)
         return 1
     return 0
 
@@ -227,7 +227,7 @@ def create_file(path, mode=None):
     -------
         The return code fo the cat(1) command.
     """
-    _logger.debug("creating file : %s" % path)
+    _logger.debug("creating file : %s", path)
     res = call([TOUCH_CMD,  path])
     if res == 0 and mode is not None:
         res = set_file_mode(path, mode)
@@ -249,5 +249,5 @@ def set_file_mode(path, mode=None):
     -------
         The return code fo the chmod(1) command.
     """
-    _logger.debug("applying mode  %s to file %s" % (mode, path))
+    _logger.debug("applying mode  %s to file %s", mode, path)
     return call([CHMOD_CMD, mode, path])

lib/oci_utils/impl/sudo_utils.py

@@ -28,12 +28,11 @@ def get_domains_name():
         All domains names as list of string.
     """
     ret = []
-    domains = sudo_utils.call_output(
-        [VIRSH_CMD, 'list', '--name', '--all']).splitlines()
-
-    for d in domains:
-        if len(d) > 0:
-            ret.append(d)
+    domains = sudo_utils.call_output([VIRSH_CMD, 'list', '--name', '--all'])
+    if domains is not None:
+        for d in domains.decode('utf-8').splitlines():
+            if len(d) > 0:
+                ret.append(d)
 
     return ret
 

lib/oci_utils/impl/virt/virt_utils.py

@@ -261,7 +261,7 @@ def _is_intf_excluded(self, interface):
         Checks if interface name, VNIC ocid or ip addr is part of excluded items
         """
 
-        for excl in self.vnic_info.get('exclude',()):
+        for excl in self.vnic_info.get('exclude', ()):
             if excl in (interface['IFACE'], interface['VNIC'], interface['ADDR']):
                 return True
         return False

lib/oci_utils/vnicutils.py

@@ -6,7 +6,6 @@ oci-public-ip = /bin/oci-public-ip
 oci-network-inspector = /bin/oci-network-inspector
 oci-network-config = /bin/oci-network-config
 oci-utils-config-helper = /usr/libexec/oci-utils-config-helper
-
 oci-image-migrate = /bin/oci-image-migrate
 oci-image-migrate-upload = /bin/oci-image-migrate-upload
 oci-image-migrate-import = /bin/oci-image-migrate-import

tests/data/properties.cfg

@@ -0,0 +1,5 @@
+[DEFAULT]
+volume-name = auto-test-volume
+volume-size = 51
+compartment-name = ImageDev
+waittime = 10

tests/data/tests/test_cli_iscsi_config/properties.cfg

@@ -84,22 +84,14 @@ def test_get_newer(self):
         -------
             No return value.
         """
-        self.assertEqual(get_newer(self.file1, self.file2), self.file2,
-                         'expected self.file2 to be the newer')
-        self.assertEqual(get_newer(self.file2, self.file1), self.file2,
-                         'expected self.file2 to be the newer')
-        self.assertEqual(get_newer(self.file1, None), self.file1,
-                         'get_newer(filename, None) != filename')
-        self.assertEqual(get_newer(None, self.file1), self.file1,
-                         'get_newer(None, filename) != filename')
-        self.assertEqual(get_newer(self.file1, self.nofile), self.file1,
-                         'get_newer(filename, \'no existing file\') != filename')
-        self.assertEqual(get_newer(self.nofile, self.file2), self.file2,
-                         'get_newer(\'no existing file\', filename, ) != filename')
-        self.assertEqual(get_newer(self.nofile, None), None,
-                         'get_newer(\'no existing file\', None) != None')
-        self.assertEqual(get_newer(None, None), None,
-                         'get_newer(None, None) != None')
+        self.assertEqual(get_newer(self.file1, self.file2), self.file2, 'expected self.file2 to be the newer')
+        self.assertEqual(get_newer(self.file2, self.file1), self.file2, 'expected self.file2 to be the newer')
+        self.assertEqual(get_newer(self.file1, None), self.file1, 'get_newer(filename, None) != filename')
+        self.assertEqual(get_newer(None, self.file1), self.file1, 'get_newer(None, filename) != filename')
+        self.assertEqual(get_newer(self.file1, self.nofile), self.file1, 'get_newer(filename, \'no existing file\') != filename')
+        self.assertEqual(get_newer(self.nofile, self.file2), self.file2, 'get_newer(\'no existing file\', filename, ) != filename')
+        self.assertEqual(get_newer(self.nofile, None), None, 'get_newer(\'no existing file\', None) != None')
+        self.assertEqual(get_newer(None, None), None, 'get_newer(None, None) != None')
 
     def test_load_cache(self):
         """
@@ -131,7 +123,7 @@ def test_load_cache(self):
 
     def test_write_cache(self):
         """
-        Test cach.write_cache().
+        Test cache.write_cache().
 
         Returns
         -------

tests/test_cache.py

@@ -3,9 +3,63 @@
 # at http://oss.oracle.com/licenses/upl.
 
 import os
+import re
 import subprocess
+import time
 import unittest
+import uuid
 from tools.oci_test_case import OciTestCase
+from tools.decorators import (skipUnlessOCI, skipUnlessRoot)
+
+def _get_volume_data(volume_data):
+    """
+    Formats the data list retrieved fr m show as a dictionary.
+
+    Parameters
+    ----------
+    volume_data: list
+
+    Returns
+    -------
+        dict: dictionary
+              { display_name : {'ocid' : OCID, 'iqn': iqn} }
+    """
+    singlespace = re.compile("\s*,\s*|\s+$")
+    volume_data_dict = dict()
+    ind = 0
+    for y in volume_data:
+        if y.startswith('Currently attached iSCSI'):
+             break
+        ind += 1
+    while ind < len(volume_data) - 1:
+        ind += 1
+        v = volume_data[ind]
+        if v.startswith('Target iqn'):
+            iqn = singlespace.sub(' ', v.split(' ')[1]).strip()
+            cnt = 0
+            ind += 1
+            for y in volume_data[ind:]:
+                if 'Volume name' in y:
+                    display_name = singlespace.sub(' ', y.split(': ', 1)[1]).strip()
+                    cnt += 1
+                elif 'Volume OCID' in y:
+                    ocid = singlespace.sub(' ', y.split(': ', 1)[1]).strip()
+                    cnt += 1
+                if cnt == 2:
+                    volume_data_dict[display_name] = {'target': iqn, 'ocid': ocid}
+                    break
+                else:
+                    ind += 1
+        if v.startswith('Volume '):
+            display_name = v.split(' ')[1]
+            ind += 1
+            for y in volume_data[ind:]:
+                if 'OCID' in y:
+                    ocid = singlespace.sub(' ', y.split(': ', 1)[1]).strip()
+                    volume_data_dict[display_name] = {'target': None, 'ocid': ocid}
+                    break
+                ind += 1
+    return volume_data_dict
 
 
 class TestCliOciIscsiConfig(OciTestCase):
@@ -25,11 +79,61 @@ def setUp(self):
         unittest.SkipTest
             If the ISCSI_CONFIG file does not exist.
         """
-        super(TestCliOciIscsiConfig, self).setUp()
+        super().setUp()
         self.iscsi_config_path = self.properties.get_property('oci-iscsi-config-path')
         if not os.path.exists(self.iscsi_config_path):
-            raise unittest.SkipTest("%s not present" %
-                                    self.iscsi_config_path)
+            raise unittest.SkipTest("%s not present" % self.iscsi_config_path)
+        self.volume_name = uuid.uuid4().hex
+        try:
+            self.waittime = int(self.properties.get_property('waittime'))
+        except Exception as e:
+            self.waittime = 10
+        try:
+            self.volume_size = self.properties.get_property('volume-size')
+        except Exception as e:
+            self.volume_size = '60'
+        try:
+            self.compartment_name = self.properties.get_property('compartment-name')
+        except Exception as e:
+            self.compartment_name = 'ImageDev'
+
+    @staticmethod
+    def _get_ocid(create_data, display_name):
+        """
+        Find the ocid of a volume.
+
+        Parameters
+        ----------
+        create_data: list
+            Block volume data.
+        display_name: str
+            Display name of the volume.
+
+        Returns
+        -------
+            str: the ocid
+        """
+        vol_dict = _get_volume_data(create_data)[display_name]
+        return vol_dict['ocid']
+
+    @staticmethod
+    def _get_iqn(create_data, display_name):
+        """
+        Find the ocid of a volume.
+
+        Parameters
+        ----------
+        create_data: list
+            Block volume data.
+        display_name: str
+            Display name of the volume.
+
+        Returns
+        -------
+            str: the ocid
+        """
+        vol_dict = _get_volume_data(create_data)[display_name]
+        return vol_dict['target']
 
     def test_display_help(self):
         """
@@ -40,10 +144,16 @@ def test_display_help(self):
             No return value.
         """
         try:
-            _ = subprocess.check_output([self.iscsi_config_path,
-                                         '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'sync', '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'usage', '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'show', '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'create', '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'attach', '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'detach', '--help'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'destroy', '--help'])
         except Exception as e:
-            self.fail('Execution has failed: %s' % str(e))
+            self.fail('oci-iscsi-config --help has failed: %s' % str(e))
 
     def test_show_no_check(self):
         """
@@ -54,10 +164,10 @@ def test_show_no_check(self):
             No return value.
         """
         try:
-            _ = subprocess.check_output([self.iscsi_config_path,
-                                         '--show'])
+            _ = subprocess.check_output([self.iscsi_config_path, '--show'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'show'])
         except Exception as e:
-            self.fail('Execution has failed: %s' % str(e))
+            self.fail('oci-iscsi-config show has failed: %s' % str(e))
 
     def test_show_all_no_check(self):
         """
@@ -68,7 +178,97 @@ def test_show_all_no_check(self):
             No return value.
         """
         try:
-            _ = subprocess.check_output([self.iscsi_config_path,
-                                         '--show', '--all'])
+            _ = subprocess.check_output([self.iscsi_config_path, '--show', '--all'])
+            _ = subprocess.check_output([self.iscsi_config_path, 'show', '--all'])
+        except Exception as e:
+            self.fail('oci-iscsi-config show --all has failed: %s' % str(e))
+
+    @skipUnlessOCI()
+    @skipUnlessRoot()
+    def test_create_destroy(self):
+        """
+        Test block volume creation and destruction
+
+        Returns
+        -------
+            No return value.
+        """
+        try:
+            create_volume_data = subprocess.check_output([self.iscsi_config_path,
+                                                          'create',
+                                                          '--size', self.volume_size,
+                                                          '--volume-name', self.volume_name,
+                                                          '--show']).decode('utf-8').splitlines()
+            # print('\nvolume created: %s' % create_volume_data)
+            time.sleep(self.waittime)
+            new_ocid = self._get_ocid(create_volume_data, self.volume_name)
+            destroy_volume_data = subprocess.check_output([self.iscsi_config_path,
+                                                           'destroy',
+                                                           '--ocids', new_ocid,
+                                                           '--yes',
+                                                           '--show']).decode('utf-8').splitlines()
+            # print('\nvolume %s destroyed: %s' % (self.volume_name, destroy_volume_data))
+        except Exception as e:
+            self.fail('oci-iscsi-config create/destroy has failed: %s' % str(e))
+
+    @skipUnlessOCI()
+    @skipUnlessRoot()
+    def test_attach_detach(self):
+        """
+        Test block device attach and detach.
+
+        Returns
+        -------
+            No return value
+        """
+        try:
+            create_volume_data = subprocess.check_output([self.iscsi_config_path,
+                                                          'create',
+                                                          '--size', self.volume_size,
+                                                          '--volume-name', self.volume_name,
+                                                          '--show']).decode('utf-8').splitlines()
+            # print('\nvolume created: %s' % create_volume_data)
+            time.sleep(self.waittime)
+            new_ocid = self._get_ocid(create_volume_data, self.volume_name)
+            attach_volume_data = subprocess.check_output([self.iscsi_config_path,
+                                                          'attach',
+                                                          '--iqns', new_ocid,
+                                                          '--show']).decode('utf-8').splitlines()
+            attach_volume_data_show = subprocess.check_output([self.iscsi_config_path,
+                                                               'show']).decode('utf-8').splitlines()
+            # print('\nvolume attached: %s' % attach_volume_data_show)
+            time.sleep(self.waittime)
+            new_iqn = self._get_iqn(attach_volume_data_show, self.volume_name)
+            detach_volume_data = subprocess.check_output([self.iscsi_config_path,
+                                                          'detach',
+                                                          '--iqns', new_iqn]).decode('utf-8').splitlines()
+            # print('\nvolume detached: %s' % detach_volume_data)
+            time.sleep(self.waittime)
+            destroy_volume_data = subprocess.check_output([self.iscsi_config_path,
+                                                           'destroy',
+                                                           '--ocids', new_ocid,
+                                                           '--yes',
+                                                           '--show']).decode('utf-8').splitlines()
+            # print('\nvolume %s destroyed: %s' % (self.volume_name, destroy_volume_data))
+        except Exception as e:
+            self.fail('oci-iscsi-config attach/detach has failed: %s' % str(e))
+
+    @skipUnlessOCI()
+    @skipUnlessRoot()
+    def test_sync(self):
+        """
+        Test volume attach.
+
+        Returns
+        -------
+            No return value.
+        """
+        try:
+            sync_data = subprocess.check_output([self.iscsi_config_path,
+                                                 'sync',
+                                                 '--apply',
+                                                 '--yes']).decode('utf-8').splitlines()
+            # print('\nvolumes synced.')
         except Exception as e:
-            self.fail('Execution has failed: %s' % str(e))
+            self.fail('oci-iscsi-config sync has failed: %s' % str(e))
+