diff --git a/qubesmgmt/app.py b/qubesmgmt/app.py index 03865af..b0bd743 100644 --- a/qubesmgmt/app.py +++ b/qubesmgmt/app.py @@ -37,7 +37,7 @@ import qubesmgmt.vm import qubesmgmt.config BUF_SIZE = 4096 - +VM_ENTRY_POINT = 'qubesmgmt.vm' class VMCollection(object): '''Collection of VMs objects''' @@ -85,7 +85,7 @@ class VMCollection(object): if item not in self: raise KeyError(item) if item not in self._vm_objects: - cls = qubesmgmt.utils.get_entry_point_one('qubesmgmt.vm', + cls = qubesmgmt.utils.get_entry_point_one(VM_ENTRY_POINT, self._vm_list[item]['class']) self._vm_objects[item] = cls(self.app, item) return self._vm_objects[item] @@ -183,6 +183,91 @@ class QubesBase(qubesmgmt.base.PropertyHolder): ''' Remove a storage pool ''' self.qubesd_call('dom0', 'mgmt.pool.Remove', name, None) + def get_label(self, label): + '''Get label as identified by index or name + + :throws KeyError: when label is not found + ''' + + # first search for name, verbatim + try: + return self.labels[label] + except KeyError: + pass + + # then search for index + if label.isdigit(): + for i in self.labels: + if i.index == int(label): + return i + + raise KeyError(label) + + @staticmethod + def get_vm_class(clsname): + '''Find the class for a domain. + + Classes are registered as setuptools' entry points in ``qubes.vm`` + group. Any package may supply their own classes. + + :param str clsname: name of the class + :return type: class + ''' + + try: + return qubesmgmt.utils.get_entry_point_one( + VM_ENTRY_POINT, clsname) + except KeyError: + raise qubesmgmt.exc.QubesException( + 'no such VM class: {!r}'.format(clsname)) + # don't catch TypeError + + def add_new_vm(self, cls, name, label, template=None, pool=None, + pools=None): + '''Create new Virtual Machine + + Example usage with custom storage pools: + + >>> app = qubesmgmt.Qubes() + >>> pools = {'private': 'external'} + >>> vm = app.add_new_vm('AppVM', 'my-new-vm', 'red', + >>> 'my-template', pools=pools) + >>> vm.netvm = app.domains['sys-whonix'] + + :param str cls: name of VM class (`AppVM`, `TemplateVM` etc) + :param str name: name of VM + :param str label: label color for new VM + :param str template: template to use (if apply for given VM class), + can be also VM object; use None for default value + :param str pool: storage pool to use instead of default one + :param dict pools: storage pool for specific volumes + :return new VM object + ''' + + if not isinstance(cls, str): + cls = cls.__name__ + + if template is not None: + template = str(template) + + if pool and pools: + raise ValueError('only one of pool= and pools= can be used') + + method_prefix = 'mgmt.vm.Create.' + payload = 'name={} label={}'.format(name, label) + if pool: + payload += ' pool={}'.format(str(pool)) + method_prefix = 'mgmt.vm.CreateInPool.' + if pools: + payload += ''.join(' pool:{}={}'.format(vol, str(pool)) + for vol, pool in sorted(pools.items())) + method_prefix = 'mgmt.vm.CreateInPool.' + + self.qubesd_call('dom0', method_prefix + cls, template, + payload.encode('utf-8')) + + return self.domains[name] + def run_service(self, dest, service, filter_esc=False, user=None, localcmd=None, **kwargs): '''Run qrexec service in a given destination diff --git a/qubesmgmt/tests/app.py b/qubesmgmt/tests/app.py index f9c52ce..7d478e4 100644 --- a/qubesmgmt/tests/app.py +++ b/qubesmgmt/tests/app.py @@ -65,4 +65,75 @@ class TC_00_VMCollection(qubesmgmt.tests.QubesTestCase): del self.app.domains['test-vm'] self.assertAllCalled() + def test_010_new_simple(self): + self.app.expected_calls[('dom0', 'mgmt.vm.Create.AppVM', None, + b'name=new-vm label=red')] = b'0\x00' + self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00new-vm class=AppVM state=Running\n' + vm = self.app.add_new_vm('AppVM', 'new-vm', 'red') + self.assertEqual(vm.name, 'new-vm') + self.assertEqual(vm.__class__.__name__, 'AppVM') + self.assertAllCalled() + def test_011_new_template(self): + self.app.expected_calls[('dom0', 'mgmt.vm.Create.TemplateVM', None, + b'name=new-template label=red')] = b'0\x00' + self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00new-template class=TemplateVM state=Running\n' + vm = self.app.add_new_vm('TemplateVM', 'new-template', 'red') + self.assertEqual(vm.name, 'new-template') + self.assertEqual(vm.__class__.__name__, 'TemplateVM') + self.assertAllCalled() + + def test_012_new_template_based(self): + self.app.expected_calls[('dom0', 'mgmt.vm.Create.AppVM', + 'some-template', b'name=new-vm label=red')] = b'0\x00' + self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00new-vm class=AppVM state=Running\n' + vm = self.app.add_new_vm('AppVM', 'new-vm', 'red', 'some-template') + self.assertEqual(vm.name, 'new-vm') + self.assertEqual(vm.__class__.__name__, 'AppVM') + self.assertAllCalled() + + def test_013_new_objects_params(self): + self.app.expected_calls[('dom0', 'mgmt.vm.Create.AppVM', + 'some-template', b'name=new-vm label=red')] = b'0\x00' + self.app.expected_calls[('dom0', 'mgmt.label.List', None, None)] = \ + b'0\x00red\nblue\n' + self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00new-vm class=AppVM state=Running\n' \ + b'some-template class=TemplateVM state=Running\n' + vm = self.app.add_new_vm(self.app.get_vm_class('AppVM'), 'new-vm', + self.app.get_label('red'), self.app.domains['some-template']) + self.assertEqual(vm.name, 'new-vm') + self.assertEqual(vm.__class__.__name__, 'AppVM') + self.assertAllCalled() + + def test_014_new_pool(self): + self.app.expected_calls[('dom0', 'mgmt.vm.CreateInPool.AppVM', None, + b'name=new-vm label=red pool=some-pool')] = b'0\x00' + self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00new-vm class=AppVM state=Running\n' + vm = self.app.add_new_vm('AppVM', 'new-vm', 'red', pool='some-pool') + self.assertEqual(vm.name, 'new-vm') + self.assertEqual(vm.__class__.__name__, 'AppVM') + self.assertAllCalled() + + def test_015_new_pools(self): + self.app.expected_calls[('dom0', 'mgmt.vm.CreateInPool.AppVM', None, + b'name=new-vm label=red pool:private=some-pool ' + b'pool:volatile=other-pool')] = b'0\x00' + self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \ + b'0\x00new-vm class=AppVM state=Running\n' + vm = self.app.add_new_vm('AppVM', 'new-vm', 'red', + pools={'private': 'some-pool', 'volatile': 'other-pool'}) + self.assertEqual(vm.name, 'new-vm') + self.assertEqual(vm.__class__.__name__, 'AppVM') + self.assertAllCalled() + + def test_020_get_label(self): + self.app.expected_calls[('dom0', 'mgmt.label.List', None, None)] = \ + b'0\x00red\nblue\n' + label = self.app.get_label('red') + self.assertEqual(label.name, 'red') + self.assertAllCalled()