feat(tests): add unit tests concerning ZFS (close xcp-ng/xcp#425)

- Check if "create" doesn't succeed without zfs packages
- Check if "scan" failed if the path is not mounted (not a ZFS mountpoint)

drivers/ZFSSR.py

@@ -58,6 +58,18 @@
 }
 
 
+def is_zfs_available():
+    import distutils.spawn
+    return distutils.spawn.find_executable('zfs') and \
+        util.pathexists('/sys/module/zfs/initstate')
+
+
+def is_zfs_path(path):
+    cmd = ['findmnt', '-o', 'FSTYPE', '-n', path]
+    fs_type = util.pread2(cmd).split('\n')[0]
+    return fs_type == 'zfs'
+
+
 class ZFSSR(FileSR.FileSR):
     DRIVER_TYPE = 'zfs'
 
@@ -66,15 +78,15 @@ def handles(type):
         return type == ZFSSR.DRIVER_TYPE
 
     def load(self, sr_uuid):
-        if not self._is_zfs_available():
+        if not is_zfs_available():
             raise xs_errors.XenError(
                 'SRUnavailable',
                 opterr='zfs is not installed or module is not loaded'
             )
         return super(ZFSSR, self).load(sr_uuid)
 
     def create(self, sr_uuid, size):
-        if not self._is_zfs_path(self.remotepath):
+        if not is_zfs_path(self.remotepath):
             raise xs_errors.XenError(
                 'ZFSSRCreate',
                 opterr='Cannot create SR, path is not a ZFS mountpoint'
@@ -90,7 +102,7 @@ def delete(self, sr_uuid):
         return super(ZFSSR, self).delete(sr_uuid)
 
     def attach(self, sr_uuid):
-        if not self._is_zfs_path(self.remotepath):
+        if not is_zfs_path(self.remotepath):
             raise xs_errors.XenError(
                 'SRUnavailable',
                 opterr='Invalid ZFS path'
@@ -106,19 +118,7 @@ def vdi(self, uuid, loadLocked=False):
     # Ensure _checkmount is overridden to prevent bad behaviors in FileSR.
     def _checkmount(self):
         return super(ZFSSR, self)._checkmount() and \
-            self._is_zfs_path(self.remotepath)
-
-    @staticmethod
-    def _is_zfs_path(path):
-        cmd = ['findmnt', '-o', 'FSTYPE', '-n', path]
-        fs_type = util.pread2(cmd).split('\n')[0]
-        return fs_type == 'zfs'
-
-    @staticmethod
-    def _is_zfs_available():
-        import distutils.spawn
-        return distutils.spawn.find_executable('zfs') and \
-            util.pathexists('/sys/module/zfs/initstate')
+            is_zfs_path(self.remotepath)
 
 
 class ZFSFileVDI(FileSR.FileVDI):

tests/test_ZFSSR.py

@@ -0,0 +1,118 @@
+import FileSR
+import mock
+import os
+import SR
+import unittest
+import ZFSSR
+
+
+XML_DEFS = os.path.dirname(os.path.abspath(__file__)) + \
+    '/../drivers/XE_SR_ERRORCODES.xml'
+
+
+class FakeZFSSR(ZFSSR.ZFSSR):
+    uuid = None
+    sr_ref = None
+    session = None
+    srcmd = None
+    other_config = {}
+    vdis = {}
+    passthrough = True
+
+    def __init__(self, srcmd, none):
+        self.dconf = srcmd.dconf
+        self.srcmd = srcmd
+
+
+class TestZFSSR(unittest.TestCase):
+    def create_zfs_sr(self, sr_uuid='asr_uuid', location='fake_path'):
+        srcmd = mock.Mock()
+        srcmd.dconf = {
+            'location': location
+        }
+        srcmd.params = {
+            'command': 'some_command',
+            'device_config': {}
+        }
+        sr = FakeZFSSR(srcmd, None)
+        sr.load(sr_uuid)
+        return sr
+
+    @mock.patch('ZFSSR.is_zfs_available', autospec=True)
+    @mock.patch('FileSR.Lock', autospec=True)
+    def test_load(self, lock, is_zfs_available):
+        self.create_zfs_sr()
+
+    @mock.patch('xs_errors.XML_DEFS', new=XML_DEFS)
+    def test_load_with_zfs_unavailable(self):
+        failed = False
+        try:
+            self.create_zfs_sr()
+        except SR.SROSError as e:
+            # Check SRUnavailable error.
+            if e.errno != 47:
+                raise
+            failed = True
+        self.assertTrue(failed)
+
+    @mock.patch('ZFSSR.is_zfs_available', autospec=True)
+    @mock.patch('ZFSSR.is_zfs_path', autospec=True)
+    @mock.patch('FileSR.Lock', autospec=True)
+    def test_create(self, lock, is_zfs_path, is_zfs_available):
+        sr = self.create_zfs_sr()
+        sr.create(sr.uuid, 42)
+
+    @mock.patch('ZFSSR.is_zfs_available', autospec=True)
+    @mock.patch('ZFSSR.is_zfs_path', autospec=True)
+    @mock.patch('FileSR.Lock', autospec=True)
+    @mock.patch('xs_errors.XML_DEFS', new=XML_DEFS)
+    def test_create_with_invalid_zfs_path(
+        self, lock, is_zfs_path, is_zfs_available
+    ):
+        failed = False
+
+        is_zfs_path.return_value = False
+        sr = self.create_zfs_sr()
+        try:
+            sr.create(sr.uuid, 42)
+        except SR.SROSError as e:
+            # Check ZFSSRCreate error.
+            if e.errno != 5000:
+                raise
+            failed = True
+        self.assertTrue(failed)
+
+    @mock.patch('ZFSSR.is_zfs_available', autospec=True)
+    @mock.patch('ZFSSR.is_zfs_path', autospec=True)
+    @mock.patch('FileSR.Lock', autospec=True)
+    @mock.patch('FileSR.FileSR._checkmount', autospec=True)
+    @mock.patch('FileSR.FileSR._loadvdis', autospec=True)
+    @mock.patch('SR.SR.scan', autospec=True)
+    @mock.patch('os.path.ismount', autospec=True)
+    def test_scan(
+        self, ismount, scan, _loadvdis, _checkmount, lock,
+        is_zfs_path, is_zfs_available
+    ):
+        sr = self.create_zfs_sr()
+        sr.scan(sr.uuid)
+
+    @mock.patch('ZFSSR.is_zfs_available', autospec=True)
+    @mock.patch('ZFSSR.is_zfs_path', autospec=True)
+    @mock.patch('FileSR.Lock', autospec=True)
+    @mock.patch('FileSR.FileSR._checkmount', autospec=True)
+    @mock.patch('xs_errors.XML_DEFS', new=XML_DEFS)
+    def test_scan_with_invalid_zfs_path(
+        self, _checkmount, lock, is_zfs_path, is_zfs_available
+    ):
+        failed = False
+
+        is_zfs_path.return_value = False
+        sr = self.create_zfs_sr()
+        try:
+            sr.scan(sr.uuid)
+        except SR.SROSError as e:
+            # Check SRUnavailable error.
+            if e.errno != 47:
+                raise
+            failed = True
+        self.assertTrue(failed)