api/admin: remove admin.vm.Clone operation
The same can be achieved with Create+volume.Clone QubesOS/qubes-issues#2622
This commit is contained in:
parent
fabd8119b4
commit
3dcd29afea
1
Makefile
1
Makefile
@ -35,7 +35,6 @@ ADMIN_API_METHODS_SIMPLE = \
|
||||
admin.property.List \
|
||||
admin.property.Reset \
|
||||
admin.property.Set \
|
||||
admin.vm.Clone \
|
||||
admin.vm.Create.AppVM \
|
||||
admin.vm.Create.DispVM \
|
||||
admin.vm.Create.StandaloneVM \
|
||||
|
@ -829,40 +829,6 @@ class QubesAdminAPI(qubes.api.AbstractQubesAPI):
|
||||
|
||||
self.app.save()
|
||||
|
||||
@qubes.api.method('admin.vm.Clone')
|
||||
@asyncio.coroutine
|
||||
def vm_clone(self, untrusted_payload):
|
||||
assert not self.arg
|
||||
|
||||
assert untrusted_payload.startswith(b'name=')
|
||||
untrusted_name = untrusted_payload[5:].decode('ascii')
|
||||
qubes.vm.validate_name(None, None, untrusted_name)
|
||||
new_name = untrusted_name
|
||||
|
||||
del untrusted_payload
|
||||
|
||||
if new_name in self.app.domains:
|
||||
raise qubes.exc.QubesValueError('Already exists')
|
||||
|
||||
self.fire_event_for_permission(new_name=new_name)
|
||||
|
||||
src_vm = self.dest
|
||||
|
||||
dst_vm = self.app.add_new_vm(src_vm.__class__, name=new_name)
|
||||
try:
|
||||
dst_vm.clone_properties(src_vm)
|
||||
dst_vm.tags.update(src_vm.tags)
|
||||
dst_vm.features.update(src_vm.features)
|
||||
#dst_vm.firewall.clone(src_vm.firewall)
|
||||
for devclass in src_vm.devices:
|
||||
for device_assignment in src_vm.devices[devclass].assignments():
|
||||
dst_vm.devices[devclass].attach(device_assignment.clone())
|
||||
yield from dst_vm.clone_disk_files(src_vm)
|
||||
except:
|
||||
del self.app.domains[dst_vm]
|
||||
raise
|
||||
self.app.save()
|
||||
|
||||
@qubes.api.method('admin.vm.device.{endpoint}.Available', endpoints=(ep.name
|
||||
for ep in pkg_resources.iter_entry_points('qubes.devices')),
|
||||
no_payload=True)
|
||||
|
@ -1236,72 +1236,6 @@ class TC_00_VMs(AdminAPITestCase):
|
||||
self.assertNotIn('test-vm2', self.app.domains)
|
||||
self.assertFalse(self.app.save.called)
|
||||
|
||||
@unittest.mock.patch('qubes.storage.Storage.clone')
|
||||
@unittest.mock.patch('qubes.storage.Storage.verify')
|
||||
def test_350_vm_clone(self, mock_verify, mock_clone):
|
||||
mock_clone.side_effect = self.dummy_coro
|
||||
mock_verify.side_effect = self.dummy_coro
|
||||
self.call_mgmt_func(b'admin.vm.Clone',
|
||||
b'test-vm1', b'', b'name=test-vm2')
|
||||
|
||||
self.assertIn('test-vm2', self.app.domains)
|
||||
vm = self.app.domains['test-vm2']
|
||||
self.assertEqual(vm.label, self.app.get_label('red'))
|
||||
self.assertEqual(vm.template, self.app.domains['test-template'])
|
||||
self.assertEqual(vm.tags, self.vm.tags)
|
||||
self.assertEqual(vm.features, self.vm.features)
|
||||
#self.assertEqual(vm.firewall, self.vm.firewall)
|
||||
self.assertEqual(mock_clone.mock_calls,
|
||||
[unittest.mock.call(self.app.domains['test-vm2']).clone(
|
||||
self.app.domains['test-vm1'])])
|
||||
self.assertTrue(os.path.exists(os.path.join(
|
||||
self.test_base_dir, 'appvms', 'test-vm2')))
|
||||
|
||||
self.assertTrue(self.app.save.called)
|
||||
|
||||
@unittest.mock.patch('qubes.storage.Storage.clone')
|
||||
@unittest.mock.patch('qubes.storage.Storage.verify')
|
||||
def test_351_vm_clone_extra_params(self, mock_verify, mock_clone):
|
||||
mock_clone.side_effect = self.dummy_coro
|
||||
mock_verify.side_effect = self.dummy_coro
|
||||
with self.assertRaises(qubes.exc.QubesException):
|
||||
self.call_mgmt_func(b'admin.vm.Clone',
|
||||
b'test-vm1', b'', b'name=test-vm2 label=red')
|
||||
|
||||
self.assertNotIn('test-vm2', self.app.domains)
|
||||
self.assertEqual(mock_clone.mock_calls, [])
|
||||
self.assertFalse(os.path.exists(os.path.join(
|
||||
self.test_base_dir, 'appvms', 'test-vm2')))
|
||||
|
||||
self.assertFalse(self.app.save.called)
|
||||
|
||||
@unittest.mock.patch('qubes.storage.Storage.clone')
|
||||
@unittest.mock.patch('qubes.storage.Storage.verify')
|
||||
def test_352_vm_clone_duplicate_name(self, mock_verify, mock_clone):
|
||||
mock_clone.side_effect = self.dummy_coro
|
||||
mock_verify.side_effect = self.dummy_coro
|
||||
with self.assertRaises(qubes.exc.QubesException):
|
||||
self.call_mgmt_func(b'admin.vm.Clone',
|
||||
b'test-vm1', b'', b'name=test-vm1')
|
||||
|
||||
self.assertFalse(self.app.save.called)
|
||||
|
||||
@unittest.mock.patch('qubes.storage.Storage.clone')
|
||||
@unittest.mock.patch('qubes.storage.Storage.verify')
|
||||
def test_353_vm_clone_invalid_name(self, mock_verify, mock_clone):
|
||||
mock_clone.side_effect = self.dummy_coro
|
||||
mock_verify.side_effect = self.dummy_coro
|
||||
with self.assertRaises(qubes.exc.QubesException):
|
||||
self.call_mgmt_func(b'admin.vm.Clone',
|
||||
b'test-vm1', b'', b'name=test-vm2/..')
|
||||
|
||||
self.assertNotIn('test-vm2/..', self.app.domains)
|
||||
self.assertEqual(mock_clone.mock_calls, [])
|
||||
self.assertFalse(os.path.exists(os.path.join(
|
||||
self.test_base_dir, 'appvms', 'test-vm2/..')))
|
||||
|
||||
self.assertFalse(self.app.save.called)
|
||||
|
||||
|
||||
def test_400_property_list(self):
|
||||
# actual function tested for admin.vm.property.* already
|
||||
|
Loading…
Reference in New Issue
Block a user