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

backup: restructude the code into classes

Browse Source 
Introduce two main classes Backup and BackupRestore for storing the
state of the desired operation. Then a simple interface to adjust
parameters.

(Almost) no functional change.

QubesOS/qubes-issues#1213
QubesOS/qubes-issues#1214
This commit is contained in:
Marek Marczykowski-Górecki 2016-03-13 13:49:42 +01:00 committed by Wojtek Porczyk
parent fd1f8def6a
commit e557fe9989
3 changed files with 1598 additions and 1598 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3114
qubes/backup.py
View File

File diff suppressed because it is too large Load Diff

36
qubes/tests/__init__.py
View File

@ -717,25 +717,22 @@ class BackupTestsMixin(SystemTestsMixin):
return vms return vms
def make_backup(self, vms, prepare_kwargs=dict(), do_kwargs=dict(), def make_backup(self, vms, target=None, expect_failure=False, **kwargs):
target=None, expect_failure=False):
if target is None: if target is None:
target = self.backupdir target = self.backupdir
try: try:
files_to_backup = \ backup = qubes.backup.Backup(self.app, vms, **kwargs)
qubes.backup.backup_prepare(self.app, vms,
print_callback=self.print_callback,
**prepare_kwargs)
except qubes.exc.QubesException as e: except qubes.exc.QubesException as e:
if not expect_failure: if not expect_failure:
self.fail("QubesException during backup_prepare: %s" % str(e)) self.fail("QubesException during backup_prepare: %s" % str(e))
else: else:
raise raise
backup.passphrase = 'qubes'
backup.target_dir = target
try: try:
qubes.backup.backup_do(self.app, target, files_to_backup, "qubes", backup.backup_do()
progress_callback=self.print_progress,
**do_kwargs)
except qubes.exc.QubesException as e: except qubes.exc.QubesException as e:
if not expect_failure: if not expect_failure:
self.fail("QubesException during backup_do: %s" % str(e)) self.fail("QubesException during backup_do: %s" % str(e))
@ -754,22 +751,17 @@ class BackupTestsMixin(SystemTestsMixin):
backupfile = source backupfile = source
with self.assertNotRaises(qubes.exc.QubesException): with self.assertNotRaises(qubes.exc.QubesException):
backup_info = qubes.backup.backup_restore_prepare( restore_op = qubes.backup.BackupRestore(
backupfile, "qubes", self.app, backupfile, appvm, "qubes")
host_collection=self.app, if options:
print_callback=self.print_callback, for key, value in options.iteritems():
appvm=appvm, setattr(restore_op.options, key, value)
options=options or {}) restore_info = restore_op.get_restore_info()
if self.verbose: if self.verbose:
qubes.backup.backup_restore_print_summary(backup_info) print restore_op.get_restore_summary(restore_info)
with self.assertNotRaises(qubes.exc.QubesException): with self.assertNotRaises(qubes.exc.QubesException):
qubes.backup.backup_restore_do( restore_op.restore_do(restore_info)
backup_info,
host_collection=self.app,
print_callback=self.print_callback if self.verbose else None,
error_callback=self.error_callback)
# maybe someone forgot to call .save() # maybe someone forgot to call .save()
self.reload_db() self.reload_db()

46
qubes/tests/int/backup.py
View File

@ -45,7 +45,7 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def test_001_compressed_backup(self): def test_001_compressed_backup(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
self.make_backup(vms, do_kwargs={'compressed': True}) self.make_backup(vms, compressed=True)
self.remove_vms(vms) self.remove_vms(vms)
self.restore_backup() self.restore_backup()
for vm in vms: for vm in vms:
@ -53,7 +53,7 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def test_002_encrypted_backup(self): def test_002_encrypted_backup(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
self.make_backup(vms, do_kwargs={'encrypted': True}) self.make_backup(vms, encrypted=True)
self.remove_vms(vms) self.remove_vms(vms)
self.restore_backup() self.restore_backup()
for vm in vms: for vm in vms:
@ -61,10 +61,7 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def test_003_compressed_encrypted_backup(self): def test_003_compressed_encrypted_backup(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
self.make_backup(vms, self.make_backup(vms, compressed=True, encrypted=True)
do_kwargs={
'compressed': True,
'encrypted': True})
self.remove_vms(vms) self.remove_vms(vms)
self.restore_backup() self.restore_backup()
for vm in vms: for vm in vms:
@ -96,7 +93,7 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def test_005_compressed_custom(self): def test_005_compressed_custom(self):
vms = self.create_backup_vms() vms = self.create_backup_vms()
self.make_backup(vms, do_kwargs={'compressed': "bzip2"}) self.make_backup(vms, compressed="bzip2")
self.remove_vms(vms) self.remove_vms(vms)
self.restore_backup() self.restore_backup()
for vm in vms: for vm in vms:
@ -135,16 +132,16 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
vms = self.create_backup_vms() vms = self.create_backup_vms()
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 in vms:
with self.assertNotRaises(qubes.exc.QubesVMNotFoundError): with self.assertNotRaises(
(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 vm.netvm and not vm.property_is_default('netvm'):
self.assertEqual(restored_vm.netvm.name, vm.netvm.name + '1') self.assertEqual(restored_vm.netvm.name, vm.netvm.name + '1')
class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin): class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin):
def setUp(self): def setUp(self):
super(TC_10_BackupVMMixin, self).setUp() super(TC_10_BackupVMMixin, self).setUp()
@ -160,12 +157,9 @@ class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin):
vms = self.create_backup_vms() vms = self.create_backup_vms()
self.backupvm.start() self.backupvm.start()
self.backupvm.run("mkdir '/var/tmp/backup directory'", wait=True) self.backupvm.run("mkdir '/var/tmp/backup directory'", wait=True)
self.make_backup(vms, self.make_backup(vms, target_vm=self.backupvm,
do_kwargs={ compressed=True, encrypted=True,
'appvm': self.backupvm, target='/var/tmp/backup directory')
'compressed': True,
'encrypted': True},
target='/var/tmp/backup directory')
self.remove_vms(vms) self.remove_vms(vms)
p = self.backupvm.run("ls /var/tmp/backup*/qubes-backup*", p = self.backupvm.run("ls /var/tmp/backup*/qubes-backup*",
passio_popen=True) passio_popen=True)
@ -177,12 +171,9 @@ class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin):
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()
self.backupvm.start() self.backupvm.start()
self.make_backup(vms, self.make_backup(vms, target_vm=self.backupvm,
do_kwargs={ compressed=True, encrypted=True,
'appvm': self.backupvm, target='dd of=/var/tmp/backup-test')
'compressed': True,
'encrypted': True},
target='dd of=/var/tmp/backup-test')
self.remove_vms(vms) self.remove_vms(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)
@ -205,13 +196,10 @@ class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin):
if retcode != 0: if retcode != 0:
raise RuntimeError("Failed to prepare backup directory") raise RuntimeError("Failed to prepare backup directory")
with self.assertRaises(qubes.exc.QubesException): with self.assertRaises(qubes.exc.QubesException):
self.make_backup(vms, self.make_backup(vms, target_vm=self.backupvm,
do_kwargs={ compressed=False, encrypted=True,
'appvm': self.backupvm, target='/home/user/backup',
'compressed': False, expect_failure=True)
'encrypted': True},
target='/home/user/backup',
expect_failure=True)
def load_tests(loader, tests, pattern): def load_tests(loader, tests, pattern):