drakrun: VM unit test and refactor

drakrun/drakrun/storage.py

@@ -66,6 +66,10 @@ def delete_vm_volume(self, vm_id: int):
         """ Delete vm_id disk volume """
         raise NotImplementedError
 
+    def exists_vm(self, vm_id: int):
+        """ Checks if vm_id disk volume exists """
+        raise NotImplementedError
+
 
 class ZfsStorageBackend(StorageBackendBase):
     """Implements storage backend based on ZFS zvols"""
@@ -193,6 +197,10 @@ def delete_zfs_tank(self):
             logging.error(exc.stdout)
             raise Exception(f"Couldn't delete {self.zfs_tank_name}")
 
+    def exists_vm(self, vm_id: int) -> bool:
+        """ Checks if vm_id disk volume exists """
+        return os.path.exists(f'/dev/zvol/{self.zfs_tank_name}/vm-{vm_id}')
+
 
 class Qcow2StorageBackend(StorageBackendBase):
     """ Implements storage backend based on QEMU QCOW2 image format """
@@ -274,6 +282,11 @@ def delete_vm_volume(self, vm_id: str):
         if not safe_delete(disk_path):
             raise Exception(f"Couldn't delete vm-{vm_id}.img")
 
+    def exists_vm(self, vm_id: int) -> bool:
+        """ Checks if vm_id disk volume exists """
+        disk_path = os.path.join(VOLUME_DIR, f"vm-{vm_id}.img")
+        return os.path.exists(disk_path)
+
 
 class LvmStorageBackend(StorageBackendBase):
     """Implements storage backend based on lvm storage"""
@@ -431,6 +444,10 @@ def import_vm0(self, path: str):
             check=True
         )
 
+    def exists_vm(self, vm_id: int):
+        """ Checks if vm_id disk volume exists """
+        return os.path.exists(f'/dev/{self.lvm_volume_group}/vm-{vm_id}')
+
 
 REGISTERED_BACKENDS = {
     "qcow2": Qcow2StorageBackend,

drakrun/drakrun/test/common_utils.py

@@ -0,0 +1,23 @@
+import shutil
+import os
+import contextlib
+
+
+@contextlib.contextmanager
+def remove_files(paths):
+
+    # change names
+    for i in paths:
+        try:
+            shutil.move(i, f"{i}.bak")
+        except FileNotFoundError:
+            pass
+
+    yield
+
+    # restore the names
+    for i in paths:
+        try:
+            shutil.move(f"{i}.bak", i)
+        except FileNotFoundError:
+            pass

drakrun/drakrun/test/conftest.py

@@ -0,0 +1,45 @@
+from typing import Dict, Tuple
+import pytest
+
+# store history of failures per test class name and per index in parametrize (if parametrize used)
+_test_failed_incremental: Dict[str, Dict[Tuple[int, ...], str]] = {}
+
+
+def pytest_runtest_makereport(item, call):
+    if "incremental" in item.keywords:
+        # incremental marker is used
+        if call.excinfo is not None:
+            # the test has failed
+            # retrieve the class name of the test
+            cls_name = str(item.cls)
+            # retrieve the index of the test (if parametrize is used in combination with incremental)
+            parametrize_index = (
+                tuple(item.callspec.indices.values())
+                if hasattr(item, "callspec")
+                else ()
+            )
+            # retrieve the name of the test function
+            test_name = item.originalname or item.name
+            # store in _test_failed_incremental the original name of the failed test
+            _test_failed_incremental.setdefault(cls_name, {}).setdefault(
+                parametrize_index, test_name
+            )
+
+
+def pytest_runtest_setup(item):
+    if "incremental" in item.keywords:
+        # retrieve the class name of the test
+        cls_name = str(item.cls)
+        # check if a previous test has failed for this class
+        if cls_name in _test_failed_incremental:
+            # retrieve the index of the test (if parametrize is used in combination with incremental)
+            parametrize_index = (
+                tuple(item.callspec.indices.values())
+                if hasattr(item, "callspec")
+                else ()
+            )
+            # retrieve the name of the first test function to fail for this class name and index
+            test_name = _test_failed_incremental[cls_name].get(parametrize_index, None)
+            # if name found, test has failed for the combination of class name & test name
+            if test_name is not None:
+                pytest.xfail("previous test failed ({})".format(test_name))

drakrun/drakrun/test/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+markers =
+    incremental: incremental tests

drakrun/drakrun/test/test_vm.py

@@ -0,0 +1,107 @@
+import pytest
+
+from drakrun.config import InstallInfo, VM_CONFIG_DIR, VOLUME_DIR
+from drakrun.vm import (
+    VirtualMachine,
+    generate_vm_conf,
+    get_all_vm_conf,
+    delete_vm_conf
+)
+from drakrun.storage import get_storage_backend
+from drakrun.draksetup import find_default_interface
+from drakrun.networking import setup_vm_network, delete_vm_network
+from common_utils import remove_files
+import drakrun
+import subprocess
+import os
+import shutil
+import logging
+
+
+def backup(install_info):
+    pass
+
+
+def restore(install_info):
+    pass
+
+
+@pytest.fixture(autouse=True)
+def enable_logging():
+    logging.basicConfig(
+        level=logging.DEBUG,
+        format='[%(asctime)s][%(levelname)s] %(message)s',
+        handlers=[logging.StreamHandler()]
+    )
+
+
+@pytest.fixture(scope="module")
+def backend():
+    install_info = InstallInfo.try_load()
+
+    if install_info is None:
+        pytest.skip("no install info")
+    backend = get_storage_backend(install_info)
+
+    backup(install_info)
+
+    yield backend
+
+    restore(install_info)
+
+
+@pytest.mark.incremental
+class TestVM:
+    def test_vm_name(self, backend):
+        self.vm = VirtualMachine(backend, 0)
+        assert self.vm.vm_name == 'vm-0'
+
+    def test_backend(self, backend):
+        # else restore tests will fail
+        assert backend.exists_vm(0) is True
+
+    def test_vm_restore(self, backend):
+        self.vm = VirtualMachine(backend, 0)
+
+        # I think this part should be abstracted and automatically handled when creating or destroying VMs
+        setup_vm_network(0, True, find_default_interface(), '8.8.8.8')
+
+        # if snapshot doesn't exist
+        with remove_files([os.path.join(VOLUME_DIR, 'snapshot.sav')]):
+            with pytest.raises(Exception):
+                self.vm.restore()
+                assert self.vm.is_running is False
+
+        # if configuration file doesn't exist
+        with remove_files([os.path.join(VM_CONFIG_DIR, 'vm-0.cfg')]):
+            with pytest.raises(Exception):
+                self.vm.restore()
+                assert self.vm.is_running is False
+
+        # monkeypatch will be required to hide the storage backend
+        if backend.exists_vm(0) is False:
+            with pytest.raises(Exception):
+                self.vm.restore()
+                assert self.vm.is_running is False
+
+        # should not raise any exceptions if everything is fine
+        self.vm.restore()
+        assert self.vm.is_running is True
+
+        # restoring a restored VM
+        # what should be the expected behavior?
+        # self.vm.restore()
+
+        delete_vm_network(0, True, find_default_interface(), '8.8.8.8')
+
+    def test_vm_destroy(self):
+        self.vm = VirtualMachine(backend, 0)
+
+        # VM should be running from the previous test
+
+        self.vm.destroy()
+        assert self.vm.is_running is False
+
+        # should 2nd time destory raise/log exceptions and then handle it?
+        # vm.destroy()
+        # assert vm.is_running is False

drakrun/drakrun/vm.py

@@ -91,10 +91,31 @@ def is_running(self) -> bool:
         )
         return res.returncode == 0
 
+    """
+    I have a few suggestions that can help remove the raw xl commands from draksetup
+    Here is the abstract, should i pursue with the idea?
+    """
+    # def uptime(self):
+    #   xl uptime command
+
+    # def create(self):
+    #   xl create command
+
+    # def pause(self):
+    #   xl pause command
+
+    # def unpause(self):
+    #   xl unpause command
+
+    # def save(self, pause=False):
+    #   xl save command
+
     def restore(self) -> None:
         """ Restore virtual machine from snapshot.
         :raises: subprocess.CalledProcessError
         """
+        # if the vm is running
+        # shouldn't we raise exceptions? and then handle it?
         if self.is_running:
             self.destroy()
         cfg_path = Path(VM_CONFIG_DIR) / f"{self.vm_name}.cfg"