diff --git a/core/storage/xen.py b/core/storage/xen.py index 06bd0186..853f0268 100644 --- a/core/storage/xen.py +++ b/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) + diff --git a/tests/storage.py b/tests/storage.py index d016fb1a..e1b408f7 100644 --- a/tests/storage.py +++ b/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)