Add configurable pool_dir to XenPool

core/storage/xen.py

@@ -28,7 +28,9 @@ import re
 import subprocess
 import sys
 
-from qubes.qubes import QubesException, defaults, vm_files
+from qubes.qubes import (QubesAdminVm, QubesAppVm, QubesException, QubesHVm,
+                         QubesNetVm, QubesProxyVm, QubesTemplateHVm,
+                         QubesTemplateVm, defaults, vm_files)
 from qubes.storage import Pool, QubesVmStorage
 
 
@@ -37,7 +39,16 @@ class XenStorage(QubesVmStorage):
     Class for VM storage of Xen VMs.
     """
 
-    def __init__(self, vm, **kwargs):
+    def __init__(self, vm, vmdir, **kwargs):
+        """ Instantiate the storage.
+
+            Args:
+                vm: a QubesVM
+                vmdir: the root directory of the pool
+        """
+        assert vm is not None
+        assert vmdir is not None
+
         super(XenStorage, self).__init__(vm, **kwargs)
 
         self.root_dev = "xvda"
@@ -45,8 +56,11 @@ class XenStorage(QubesVmStorage):
         self.volatile_dev = "xvdc"
         self.modules_dev = "xvdd"
 
+        self.vmdir = vmdir
+
         if self.vm.is_template():
-            self.rootcow_img = os.path.join(self.vmdir, vm_files["rootcow_img"])
+            self.rootcow_img = os.path.join(self.vmdir,
+                                            vm_files["rootcow_img"])
         else:
             self.rootcow_img = None
 
@@ -252,12 +266,51 @@ class XenStorage(QubesVmStorage):
 
 
 class XenPool(Pool):
+
     def __init__(self, vm, dir):
         assert vm is not None
         assert dir is not None
 
+        if not os.path.exists(dir):
+            os.mkdir(dir)
+
+        self.vmdir = self._vmdir_path(vm, dir)
         self.vm = vm
         self.dir = dir
 
     def getStorage(self):
-        return defaults['storage_class'](self.vm)
+        """ Returns an instantiated ``XenStorage``. """
+        return defaults['storage_class'](self.vm, vmdir=self.vmdir)
+
+    def _vmdir_path(self, vm, pool_dir):
+        """ Get the vm dir depending on the type of the VM.
+
+            The default QubesOS file storage saves the vm images in three
+            different directories depending on the ``QubesVM`` type:
+
+            * ``appvms`` for ``QubesAppVm`` or ``QubesHvm``
+            * ``vm-templates`` for ``QubesTemplateVm`` or ``QubesTemplateHvm``
+            * ``servicevms`` for ``QubesProxyVm``, ``QubesNetVm`` or
+                ``QubesAdminVm``
+
+            Args:
+                vm: a QubesVM
+                pool_dir: the root directory of the pool
+
+            Returns:
+                string (str) absolute path to the directory where the vm files
+                             are stored
+        """
+        vm_type = type(vm)
+
+        if vm_type in [QubesAppVm, QubesHVm]:
+            subdir = 'appvms'
+        elif vm_type in [QubesTemplateVm, QubesTemplateHVm]:
+            subdir = 'vm-templates'
+        elif vm_type in [QubesAdminVm, QubesNetVm, QubesProxyVm]:
+            subdir = 'servicevms'
+        else:
+            raise QubesException(str(vm_type) + ' unknown vm type')
+
+        return os.path.join(pool_dir, subdir, vm.name)
+

tests/storage.py

@@ -17,11 +17,13 @@
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
 
+import os
+import shutil
+
 import qubes.storage
 from qubes.qubes import defaults
-from qubes.tests import QubesTestCase, SystemTestsMixin
-
 from qubes.storage.xen import XenPool, XenStorage
+from qubes.tests import QubesTestCase, SystemTestsMixin
 
 
 class TC_00_Storage(SystemTestsMixin, QubesTestCase):
@@ -103,3 +105,118 @@ class TC_00_Pool(SystemTestsMixin, QubesTestCase):
         template = self.qc.get_default_template()
         return self.qc.add_new_vm('QubesAppVm', name=vmname, template=template,
                                   pool_name='default')
+
+
+class TC_01_Pool(SystemTestsMixin, QubesTestCase):
+
+    """ Test the paths for the default Xen file based storage
+        (``QubesXenVmStorage``).
+    """
+
+    POOL_DIR = '/var/lib/qubes/test-pool'
+    APPVMS_DIR = '/var/lib/qubes/test-pool/appvms'
+    TEMPLATES_DIR = '/var/lib/qubes/test-pool/vm-templates'
+    SERVICE_DIR = '/var/lib/qubes/test-pool/servicevms'
+
+    def setUp(self):
+        """ Add a test file based storage pool """
+        super(TC_01_Pool, self).setUp()
+        qubes.storage.add_pool('test-pool', type='xen', dir=self.POOL_DIR)
+
+    def tearDown(self):
+        """ Remove the file based storage pool after testing """
+        super(TC_01_Pool, self).tearDown()
+        qubes.storage.remove_pool("test-pool")
+        shutil.rmtree(self.POOL_DIR, ignore_errors=True)
+
+    def test_001_pool_exists(self):
+        """ Check if the storage pool was added to the storage pool config """
+        self.assertTrue(qubes.storage.pool_exists('test-pool'))
+
+    def test_002_pool_dir_create(self):
+        """ Check if the storage pool dir was created """
+
+        # The dir should not exists before
+        self.assertFalse(os.path.exists(self.POOL_DIR))
+
+        vmname = self.make_vm_name('appvm')
+        template = self.qc.get_default_template()
+        self.qc.add_new_vm('QubesAppVm', name=vmname, template=template,
+                           pool_name='test-pool')
+
+        self.assertTrue(os.path.exists(self.POOL_DIR))
+
+    def test_003_pool_dir(self):
+        """ Check if the vm storage pool_dir is the same as specified """
+        vmname = self.make_vm_name('appvm')
+        template = self.qc.get_default_template()
+        vm = self.qc.add_new_vm('QubesAppVm', name=vmname, template=template,
+                                pool_name='test-pool')
+        result = qubes.storage.get_pool('test-pool', vm).dir
+        self.assertEquals(self.POOL_DIR, result)
+
+    def test_004_app_vmdir(self):
+        """ Check the vm storage dir for an AppVm"""
+        vmname = self.make_vm_name('appvm')
+        template = self.qc.get_default_template()
+        vm = self.qc.add_new_vm('QubesAppVm', name=vmname, template=template,
+                                pool_name='test-pool')
+
+        expected = os.path.join(self.APPVMS_DIR, vm.name)
+        result = vm.storage.vmdir
+        self.assertEquals(expected, result)
+
+    def test_005_hvm_vmdir(self):
+        """ Check the vm storage dir for a HVM"""
+        vmname = self.make_vm_name('hvm')
+        vm = self.qc.add_new_vm('QubesHVm', name=vmname,
+                                pool_name='test-pool')
+
+        expected = os.path.join(self.APPVMS_DIR, vm.name)
+        result = vm.storage.vmdir
+        self.assertEquals(expected, result)
+
+    def test_006_net_vmdir(self):
+        """ Check the vm storage dir for a Netvm"""
+        vmname = self.make_vm_name('hvm')
+        vm = self.qc.add_new_vm('QubesNetVm', name=vmname,
+                                pool_name='test-pool')
+
+        expected = os.path.join(self.SERVICE_DIR, vm.name)
+        result = vm.storage.vmdir
+        self.assertEquals(expected, result)
+
+    def test_007_proxy_vmdir(self):
+        """ Check the vm storage dir for a ProxyVm"""
+        vmname = self.make_vm_name('proxyvm')
+        vm = self.qc.add_new_vm('QubesProxyVm', name=vmname,
+                                pool_name='test-pool')
+
+        expected = os.path.join(self.SERVICE_DIR, vm.name)
+        result = vm.storage.vmdir
+        self.assertEquals(expected, result)
+
+    def test_008_admin_vmdir(self):
+        """ Check the vm storage dir for a AdminVm"""
+        # TODO How to test AdminVm?
+        pass
+
+    def test_009_template_vmdir(self):
+        """ Check the vm storage dir for a TemplateVm"""
+        vmname = self.make_vm_name('templatevm')
+        vm = self.qc.add_new_vm('QubesTemplateVm', name=vmname,
+                                pool_name='test-pool')
+
+        expected = os.path.join(self.TEMPLATES_DIR, vm.name)
+        result = vm.storage.vmdir
+        self.assertEquals(expected, result)
+
+    def test_010_template_hvm_vmdir(self):
+        """ Check the vm storage dir for a TemplateHVm"""
+        vmname = self.make_vm_name('templatehvm')
+        vm = self.qc.add_new_vm('QubesTemplateHVm', name=vmname,
+                                pool_name='test-pool')
+
+        expected = os.path.join(self.TEMPLATES_DIR, vm.name)
+        result = vm.storage.vmdir
+        self.assertEquals(expected, result)