Qubes/core-admin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests/backups: convert to new restore handling - using qubesadmin module

Browse Source 
Besides converting itself, change how the test verify restore
correctness: first collect VM metadata (and hashes of data) into plain
dict, then compare against it. This allow to destroy old VMs objects
before restoring the backup, so avoid having duplicate objects of the
same VM - which results in weird effects like trying to undefine libvirt
object twice.
This commit is contained in:
Marek Marczykowski-Górecki 2017-10-21 03:33:29 +02:00
parent 59cebd5439
commit 31a55dcd18
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
1 changed files with 289 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

305
qubes/tests/integ/backup.py
View File

@ -27,6 +27,9 @@ import os
import shutil import shutil
import sys import sys
import asyncio
import qubes import qubes
import qubes.backup import qubes.backup
import qubes.exc import qubes.exc
@ -38,6 +41,14 @@ import qubes.vm.appvm
import qubes.vm.templatevm import qubes.vm.templatevm
import qubes.vm.qubesvm import qubes.vm.qubesvm
try:
import qubesadmin.backup.restore
import qubesadmin.exc
restore_available = True
except ImportError:
restore_available = False
# noinspection PyAttributeOutsideInit # noinspection PyAttributeOutsideInit
class BackupTestsMixin(object): class BackupTestsMixin(object):
class BackupErrorHandler(logging.Handler): class BackupErrorHandler(logging.Handler):
@ -65,15 +76,15 @@ class BackupTestsMixin(object):
self.error_handler = self.BackupErrorHandler(self.error_detected, self.error_handler = self.BackupErrorHandler(self.error_detected,
level=logging.WARNING) level=logging.WARNING)
backup_log = logging.getLogger('qubes.backup') backup_log = logging.getLogger('qubesadmin.backup')
backup_log.addHandler(self.error_handler) backup_log.addHandler(self.error_handler)
def tearDown(self): def tearDown(self):
super(BackupTestsMixin, self).tearDown()
shutil.rmtree(self.backupdir) shutil.rmtree(self.backupdir)
backup_log = logging.getLogger('qubes.backup') backup_log = logging.getLogger('qubes.backup')
backup_log.removeHandler(self.error_handler) backup_log.removeHandler(self.error_handler)
super(BackupTestsMixin, self).tearDown()
def fill_image(self, path, size=None, sparse=False): def fill_image(self, path, size=None, sparse=False):
block_size = 4096 block_size = 4096
@ -112,7 +123,6 @@ class BackupTestsMixin(object):
self.log.debug("Creating %s" % vmname) self.log.debug("Creating %s" % vmname)
testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM, testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=vmname, template=template, label='red') name=vmname, template=template, label='red')
testvm1.uses_default_netvm = False
testvm1.netvm = testnet testvm1.netvm = testnet
self.loop.run_until_complete( self.loop.run_until_complete(
testvm1.create_on_disk(pool=pool)) testvm1.create_on_disk(pool=pool))
@ -175,10 +185,18 @@ class BackupTestsMixin(object):
else: else:
raise raise
def remove_vms(self, vms):
vms = list(vms)
for vm in vms:
vm.netvm = None
vm.default_dispvm = None
super(BackupTestsMixin, self).remove_vms(vms)
def restore_backup(self, source=None, appvm=None, options=None, def restore_backup(self, source=None, appvm=None, options=None,
expect_errors=None, manipulate_restore_info=None, expect_errors=None, manipulate_restore_info=None,
passphrase='qubes'): passphrase='qubes'):
self.skipTest('Test not converted to Qubes 4.0') if not restore_available:
self.skipTest('qubesadmin module not available')
if source is None: if source is None:
backupfile = os.path.join(self.backupdir, backupfile = os.path.join(self.backupdir,
@ -186,19 +204,30 @@ class BackupTestsMixin(object):
else: else:
backupfile = source backupfile = source
with self.assertNotRaises(qubes.exc.QubesException): client_app = qubesadmin.Qubes()
restore_op = qubes.backup.BackupRestore( if appvm:
self.app, backupfile, appvm, passphrase) appvm = self.loop.run_until_complete(
self.loop.run_in_executor(None,
client_app.domains.__getitem__, appvm.name))
with self.assertNotRaises(qubesadmin.exc.QubesException):
restore_op = self.loop.run_until_complete(
self.loop.run_in_executor(None,
qubesadmin.backup.restore.BackupRestore,
client_app, backupfile, appvm, passphrase))
if options: if options:
for key, value in options.items(): for key, value in options.items():
setattr(restore_op.options, key, value) setattr(restore_op.options, key, value)
restore_info = restore_op.get_restore_info() restore_info = self.loop.run_until_complete(
self.loop.run_in_executor(None,
restore_op.get_restore_info))
if callable(manipulate_restore_info): if callable(manipulate_restore_info):
restore_info = manipulate_restore_info(restore_info) restore_info = manipulate_restore_info(restore_info)
self.log.debug(restore_op.get_restore_summary(restore_info)) self.log.debug(restore_op.get_restore_summary(restore_info))
with self.assertNotRaises(qubes.exc.QubesException): with self.assertNotRaises(qubesadmin.exc.QubesException):
restore_op.restore_do(restore_info) self.loop.run_until_complete(
self.loop.run_in_executor(None,
restore_op.restore_do, restore_info))
errors = [] errors = []
if expect_errors is None: if expect_errors is None:
@ -230,7 +259,7 @@ class BackupTestsMixin(object):
for name, volume in vm.volumes.items(): for name, volume in vm.volumes.items():
if not volume.rw or not volume.save_on_stop: if not volume.rw or not volume.save_on_stop:
continue continue
vol_path = vm.storage.get_pool(volume).export(volume) vol_path = volume.export()
hasher = hashlib.sha1() hasher = hashlib.sha1()
with open(vol_path, 'rb') as afile: with open(vol_path, 'rb') as afile:
for buf in iter(lambda: afile.read(4096000), b''): for buf in iter(lambda: afile.read(4096000), b''):
@ -238,7 +267,37 @@ class BackupTestsMixin(object):
hashes[vm.name][name] = hasher.hexdigest() hashes[vm.name][name] = hasher.hexdigest()
return hashes return hashes
def assertCorrectlyRestored(self, orig_vms, orig_hashes): def get_vms_info(self, vms):
''' Get VM metadata, for comparing VM later without holding actual
reference to the old object.'''
vms_info = {}
for vm in vms:
vm_info = {
'properties': {},
'default': {},
'devices': {},
}
for prop in ('name', 'kernel',
'memory', 'maxmem', 'kernelopts',
'services', 'vcpus', 'features'
'include_in_backups', 'default_user', 'qrexec_timeout',
'autostart', 'pci_strictreset', 'debug',
'internal', 'netvm', 'template', 'label'):
if not hasattr(vm, prop):
continue
vm_info['properties'][prop] = str(getattr(vm, prop))
vm_info['default'][prop] = vm.property_is_default(prop)
for dev_class in vm.devices.keys():
vm_info['devices'][dev_class] = {}
for dev_ass in vm.devices[dev_class].assignments():
vm_info['devices'][dev_class][str(dev_ass.device)] = \
dev_ass.options
vms_info[vm.name] = vm_info
return vms_info
def assertCorrectlyRestored(self, vms_info, orig_hashes):
''' Verify if restored VMs are identical to those before backup. ''' Verify if restored VMs are identical to those before backup.
:param orig_vms: collection of original QubesVM objects :param orig_vms: collection of original QubesVM objects
@ -246,114 +305,108 @@ class BackupTestsMixin(object):
before backup before backup
:return: :return:
''' '''
for vm in orig_vms: for vm_name in vms_info:
self.assertIn(vm.name, self.app.domains) vm_info = vms_info[vm_name]
restored_vm = self.app.domains[vm.name] self.assertIn(vm_name, self.app.domains)
for prop in ('name', 'kernel', restored_vm = self.app.domains[vm_name]
'memory', 'maxmem', 'kernelopts', for prop in vm_info['properties']:
'services', 'vcpus', 'features'
'include_in_backups', 'default_user', 'qrexec_timeout',
'autostart', 'pci_strictreset', 'debug',
'internal'):
if not hasattr(vm, prop):
continue
self.assertEqual( self.assertEqual(
getattr(vm, prop), getattr(restored_vm, prop), vm_info['properties'][prop],
str(getattr(restored_vm, prop)),
"VM {} - property {} not properly restored".format( "VM {} - property {} not properly restored".format(
vm.name, prop)) vm_name, prop))
for prop in ('netvm', 'template', 'label'): self.assertEqual(
if not hasattr(vm, prop): vm_info['default'][prop],
continue restored_vm.property_is_default(prop),
orig_value = getattr(vm, prop) "VM {} - property {} differs in being default".format(
restored_value = getattr(restored_vm, prop) vm_name, prop))
if orig_value and restored_value: for dev_class in vm_info['devices']:
self.assertEqual(orig_value.name, restored_value.name, for dev in vm_info['devices'][dev_class]:
"VM {} - property {} not properly restored".format( found = False
vm.name, prop)) for restored_dev_ass in restored_vm.devices[
else: dev_class].assignments():
self.assertEqual(orig_value, restored_value, if str(restored_dev_ass.device) == dev:
"VM {} - property {} not properly restored".format( found = True
vm.name, prop)) self.assertEqual(vm_info['devices'][dev_class][dev],
for dev_class in vm.devices.keys(): restored_dev_ass.options,
for dev in vm.devices[dev_class]: 'VM {} - {} device {} options mismatch'.format(
self.assertIn(dev, restored_vm.devices[dev_class], vm_name, dev_class, str(dev)))
"VM {} - {} device not restored".format( self.assertTrue(found,
vm.name, dev_class)) 'VM {} - {} device {} not restored'.format(
vm_name, dev_class, dev))
if orig_hashes: if orig_hashes:
hashes = self.vm_checksum([restored_vm])[restored_vm.name] hashes = self.vm_checksum([restored_vm])[restored_vm.name]
self.assertEqual(orig_hashes[vm.name], hashes, self.assertEqual(orig_hashes[vm_name], hashes,
"VM {} - disk images are not properly restored".format( "VM {} - disk images are not properly restored".format(
vm.name)) vm_name))
class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase): class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase):
def test_000_basic_backup(self): def test_000_basic_backup(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms) self.make_backup(vms)
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
finally:
del vms
self.restore_backup() self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
self.remove_vms(reversed(vms))
def test_001_compressed_backup(self): def test_001_compressed_backup(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms, compressed=True) self.make_backup(vms, compressed=True)
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
finally:
del vms
self.restore_backup() self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
def test_002_encrypted_backup(self):
vms = self.create_backup_vms()
orig_hashes = self.vm_checksum(vms)
self.make_backup(vms, encrypted=True)
self.remove_vms(reversed(vms))
self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes)
def test_003_compressed_encrypted_backup(self):
vms = self.create_backup_vms()
orig_hashes = self.vm_checksum(vms)
self.make_backup(vms, compressed=True, encrypted=True)
self.remove_vms(reversed(vms))
self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes)
def test_004_sparse_multipart(self): def test_004_sparse_multipart(self):
vms = [] vms = []
try:
vmname = self.make_vm_name('testhvm2') vmname = self.make_vm_name('testhvm2')
self.log.debug("Creating %s" % vmname) self.log.debug("Creating %s" % vmname)
hvmtemplate = self.app.add_new_vm( self.hvmtemplate = self.app.add_new_vm(
qubes.vm.templatevm.TemplateVM, name=vmname, virt_mode='hvm', label='red') qubes.vm.templatevm.TemplateVM, name=vmname, virt_mode='hvm', label='red')
hvmtemplate.create_on_disk() self.loop.run_until_complete(self.hvmtemplate.create_on_disk())
self.fill_image( self.fill_image(
os.path.join(hvmtemplate.dir_path, '00file'), os.path.join(self.hvmtemplate.dir_path, '00file'),
195 * 1024 * 1024 - 4096 * 3) 195 * 1024 * 1024 - 4096 * 3)
self.fill_image(hvmtemplate.storage.export('private'), self.fill_image(self.hvmtemplate.storage.export('private'),
195 * 1024 * 1024 - 4096 * 3) 195 * 1024 * 1024 - 4096 * 3)
self.fill_image(hvmtemplate.storage.export('root'), 1024 * 1024 * 1024, self.fill_image(self.hvmtemplate.storage.export('root'), 1024 * 1024 * 1024,
sparse=True) sparse=True)
vms.append(hvmtemplate) vms.append(self.hvmtemplate)
self.app.save() self.app.save()
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms) self.make_backup(vms)
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup() self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
# TODO check vm.backup_timestamp # TODO check vm.backup_timestamp
finally:
del vms
def test_005_compressed_custom(self): def test_005_compressed_custom(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms, compression_filter="bzip2") self.make_backup(vms, compression_filter="bzip2")
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup() self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
finally:
del vms
def test_010_selective_restore(self): def test_010_selective_restore(self):
# create backup with internal dependencies (template, netvm etc) # create backup with internal dependencies (template, netvm etc)
@ -368,27 +421,39 @@ class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase):
restore_info.pop(name) restore_info.pop(name)
return restore_info return restore_info
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms, compression_filter="bzip2") self.make_backup(vms, compression_filter="bzip2")
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup(manipulate_restore_info=exclude_some) self.restore_backup(manipulate_restore_info=exclude_some)
for vm in vms: for vm_name in vms_info:
if vm.name == self.make_vm_name('test1'): if vm_name == self.make_vm_name('test1'):
# netvm was set to 'test-inst-test-net' - excluded # netvm was set to 'test-inst-test-net' - excluded
vm.netvm = qubes.property.DEFAULT vms_info[vm_name]['properties']['netvm'] = \
elif vm.name == self.make_vm_name('custom'): str(self.app.default_netvm)
vms_info[vm_name]['default']['netvm'] = True
elif vm_name == self.make_vm_name('custom'):
# template was set to 'test-inst-template' - excluded # template was set to 'test-inst-template' - excluded
vm.template = self.app.default_template vms_info[vm_name]['properties']['template'] = \
vms = [vm for vm in vms if vm.name not in exclude] str(self.app.default_template)
self.assertCorrectlyRestored(vms, orig_hashes) for excluded in exclude:
vms_info.pop(excluded, None)
self.assertCorrectlyRestored(vms_info, orig_hashes)
finally:
del vms
def test_020_encrypted_backup_non_ascii(self): def test_020_encrypted_backup_non_ascii(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
self.make_backup(vms, encrypted=True, passphrase=u'zażółć gęślą jaźń') vms_info = self.get_vms_info(vms)
self.make_backup(vms, passphrase=u'zażółć gęślą jaźń')
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup(passphrase=u'zażółć gęślą jaźń') self.restore_backup(passphrase=u'zażółć gęślą jaźń')
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
finally:
del vms
def test_100_backup_dom0_no_restore(self): def test_100_backup_dom0_no_restore(self):
# do not write it into dom0 home itself... # do not write it into dom0 home itself...
@ -403,19 +468,28 @@ class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase):
:return: :return:
""" """
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms) self.make_backup(vms)
self.remove_vms(reversed(vms))
test_dir = vms[0].dir_path test_dir = vms[0].dir_path
self.remove_vms(reversed(vms))
os.mkdir(test_dir) os.mkdir(test_dir)
with open(os.path.join(test_dir, 'some-file.txt'), 'w') as f: with open(os.path.join(test_dir, 'some-file.txt'), 'w') as f:
f.write('test file\n') f.write('test file\n')
self.restore_backup( self.restore_backup(expect_errors=[
expect_errors=[ 'Error restoring VM test-inst-test-net, skipping: Got empty '
'*** Directory {} already exists! It has been moved'.format( 'response from qubesd. See journalctl in dom0 for details.',
test_dir) 'Error setting test-inst-test1.netvm to test-inst-test-net: '
'\'"No such domain: \\\'test-inst-test-net\\\'"\'',
]) ])
self.assertCorrectlyRestored(vms, orig_hashes) del vms_info['test-inst-test-net']
vms_info['test-inst-test1']['properties']['netvm'] = \
str(self.app.default_netvm)
vms_info['test-inst-test1']['default']['netvm'] = True
self.assertCorrectlyRestored(vms_info, orig_hashes)
finally:
del vms
def test_210_auto_rename(self): def test_210_auto_rename(self):
""" """
@ -423,16 +497,22 @@ class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase):
:return: :return:
""" """
vms = self.create_backup_vms() vms = self.create_backup_vms()
vms_info = self.get_vms_info(vms)
try:
self.make_backup(vms) self.make_backup(vms)
self.restore_backup(options={ self.restore_backup(options={
'rename_conflicting': True 'rename_conflicting': True
}) })
for vm in vms: for vm_name in vms_info:
with self.assertNotRaises( with self.assertNotRaises(
(qubes.exc.QubesVMNotFoundError, KeyError)): (qubes.exc.QubesVMNotFoundError, KeyError)):
restored_vm = self.app.domains[vm.name + '1'] restored_vm = self.app.domains[vm_name + '1']
if vm.netvm and not vm.property_is_default('netvm'): if vms_info[vm_name]['properties']['netvm'] and \
self.assertEqual(restored_vm.netvm.name, vm.netvm.name + '1') not vms_info[vm_name]['default']['netvm']:
self.assertEqual(restored_vm.netvm.name,
vms_info[vm_name]['properties']['netvm'] + '1')
finally:
del vms
def _find_pool(self, volume_group, thin_pool): def _find_pool(self, volume_group, thin_pool):
''' Returns the pool matching the specified ``volume_group`` & ''' Returns the pool matching the specified ``volume_group`` &
@ -456,12 +536,15 @@ class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase):
**qubes.tests.storage_lvm.POOL_CONF) **qubes.tests.storage_lvm.POOL_CONF)
self.created_pool = True self.created_pool = True
vms = self.create_backup_vms(pool=self.pool) vms = self.create_backup_vms(pool=self.pool)
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms) self.make_backup(vms)
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup() self.restore_backup()
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
self.remove_vms(reversed(vms)) finally:
del vms
@qubes.tests.storage_lvm.skipUnlessLvmPoolExists @qubes.tests.storage_lvm.skipUnlessLvmPoolExists
def test_301_restore_to_lvm(self): def test_301_restore_to_lvm(self):
@ -473,17 +556,20 @@ class TC_00_Backup(BackupTestsMixin, qubes.tests.SystemTestCase):
**qubes.tests.storage_lvm.POOL_CONF) **qubes.tests.storage_lvm.POOL_CONF)
self.created_pool = True self.created_pool = True
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
orig_hashes = self.vm_checksum(vms) orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
self.make_backup(vms) self.make_backup(vms)
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup(options={'override_pool': self.pool.name}) self.restore_backup(options={'override_pool': self.pool.name})
self.assertCorrectlyRestored(vms, orig_hashes) self.assertCorrectlyRestored(vms_info, orig_hashes)
for vm in vms: for vm_name in vms_info:
vm = self.app.domains[vm.name] vm = self.app.domains[vm_name]
for volume in vm.volumes.values(): for volume in vm.volumes.values():
if volume.save_on_stop: if volume.save_on_stop:
self.assertEqual(volume.pool, self.pool.name) self.assertEqual(volume.pool, self.pool.name)
self.remove_vms(reversed(vms)) finally:
del vms
class TC_10_BackupVMMixin(BackupTestsMixin): class TC_10_BackupVMMixin(BackupTestsMixin):
@ -499,11 +585,14 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
def test_100_send_to_vm_file_with_spaces(self): def test_100_send_to_vm_file_with_spaces(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
try:
self.loop.run_until_complete(self.backupvm.start()) self.loop.run_until_complete(self.backupvm.start())
self.loop.run_until_complete(self.backupvm.run_for_stdio( self.loop.run_until_complete(self.backupvm.run_for_stdio(
"mkdir '/var/tmp/backup directory'")) "mkdir '/var/tmp/backup directory'"))
self.make_backup(vms, target_vm=self.backupvm, self.make_backup(vms, target_vm=self.backupvm,
compressed=True, encrypted=True, compressed=True,
target='/var/tmp/backup directory') target='/var/tmp/backup directory')
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
(backup_path, _) = self.loop.run_until_complete( (backup_path, _) = self.loop.run_until_complete(
@ -511,16 +600,25 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
backup_path = backup_path.decode().strip() backup_path = backup_path.decode().strip()
self.restore_backup(source=backup_path, self.restore_backup(source=backup_path,
appvm=self.backupvm) appvm=self.backupvm)
self.assertCorrectlyRestored(vms_info, orig_hashes)
finally:
del vms
def test_110_send_to_vm_command(self): def test_110_send_to_vm_command(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
orig_hashes = self.vm_checksum(vms)
vms_info = self.get_vms_info(vms)
try:
self.loop.run_until_complete(self.backupvm.start()) self.loop.run_until_complete(self.backupvm.start())
self.make_backup(vms, target_vm=self.backupvm, self.make_backup(vms, target_vm=self.backupvm,
compressed=True, encrypted=True, compressed=True,
target='dd of=/var/tmp/backup-test') target='dd of=/var/tmp/backup-test')
self.remove_vms(reversed(vms)) self.remove_vms(reversed(vms))
self.restore_backup(source='dd if=/var/tmp/backup-test', self.restore_backup(source='dd if=/var/tmp/backup-test',
appvm=self.backupvm) appvm=self.backupvm)
self.assertCorrectlyRestored(vms_info, orig_hashes)
finally:
del vms
def test_110_send_to_vm_no_space(self): def test_110_send_to_vm_no_space(self):
""" """
@ -529,6 +627,7 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
:return: :return:
""" """
vms = self.create_backup_vms() vms = self.create_backup_vms()
try:
self.loop.run_until_complete(self.backupvm.start()) self.loop.run_until_complete(self.backupvm.start())
self.loop.run_until_complete(self.backupvm.run_for_stdio( self.loop.run_until_complete(self.backupvm.run_for_stdio(
# Debian 7 has too old losetup to handle loop-control device # Debian 7 has too old losetup to handle loop-control device
@ -541,9 +640,11 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
user="root")) user="root"))
with self.assertRaises(qubes.exc.QubesException): with self.assertRaises(qubes.exc.QubesException):
self.make_backup(vms, target_vm=self.backupvm, self.make_backup(vms, target_vm=self.backupvm,
compressed=False, encrypted=True, compressed=False,
target='/home/user/backup', target='/home/user/backup',
expect_failure=True) expect_failure=True)
finally:
del vms
def load_tests(loader, tests, pattern): def load_tests(loader, tests, pattern):
@ -551,7 +652,7 @@ def load_tests(loader, tests, pattern):
tests.addTests(loader.loadTestsFromTestCase( tests.addTests(loader.loadTestsFromTestCase(
type( type(
'TC_10_BackupVM_' + template, 'TC_10_BackupVM_' + template,
(TC_10_BackupVMMixin, qubes.tests.QubesTestCase), (TC_10_BackupVMMixin, qubes.tests.SystemTestCase),
{'template': template}))) {'template': template})))
return tests return tests