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

Merge branch 'core3-backup' into core3-devel

Browse Source
This commit is contained in:
Wojtek Porczyk 2016-04-07 13:21:19 +02:00
commit 6c2f675b5c
11 changed files with 2711 additions and 2502 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2308
core/backup.py
View File

File diff suppressed because it is too large Load Diff

3
qubes/__init__.py
View File

@ -1224,6 +1224,9 @@ class Qubes(PropertyHolder):
self.events_enabled = True
@__builtin__.property
def store(self):
return self._store
def load(self):
'''Open qubes.xml

2261
qubes/backup.py Normal file
View File

File diff suppressed because it is too large Load Diff

223
qubes/core2migration.py Normal file
View File

@ -0,0 +1,223 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Marek Marczykowski-Górecki
# <marmarek@invisiblethingslab.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
#
#
import os
import sys
import qubes
import qubes.vm.appvm
import qubes.vm.standalonevm
import qubes.vm.templatevm
import qubes.vm.adminvm
import qubes.ext.r3compatibility
import lxml.etree
import xml.parsers.expat
class AppVM(qubes.vm.appvm.AppVM):
"""core2 compatibility AppVM class, with variable dir_path"""
dir_path = qubes.property('dir_path',
default=(lambda self: self.storage.vmdir),
saver=qubes.property.dontsave,
doc="VM storage directory",
)
def is_running(self):
return False
class StandaloneVM(qubes.vm.standalonevm.StandaloneVM):
"""core2 compatibility StandaloneVM class, with variable dir_path"""
dir_path = qubes.property('dir_path',
default=(lambda self: self.storage.vmdir),
saver=qubes.property.dontsave,
doc="VM storage directory")
def is_running(self):
return False
class Core2Qubes(qubes.Qubes):
def __init__(self, store=None, load=True, **kwargs):
if store is None:
raise ValueError("store path required")
super(Core2Qubes, self).__init__(store, load, **kwargs)
def load_globals(self, element):
default_template = element.get("default_template")
self.default_template = int(default_template) \
if default_template.lower() != "none" else None
default_netvm = element.get("default_netvm")
if default_netvm is not None:
self.default_netvm = int(default_netvm) \
if default_netvm != "None" else None
default_fw_netvm = element.get("default_fw_netvm")
if default_fw_netvm is not None:
self.default_fw_netvm = int(default_fw_netvm) \
if default_fw_netvm != "None" else None
updatevm = element.get("updatevm")
if updatevm is not None:
self.updatevm = int(updatevm) \
if updatevm != "None" else None
clockvm = element.get("clockvm")
if clockvm is not None:
self.clockvm = int(clockvm) \
if clockvm != "None" else None
self.default_kernel = element.get("default_kernel")
def set_netvm_dependency(self, element):
kwargs = {}
attr_list = ("qid", "uses_default_netvm", "netvm_qid")
for attribute in attr_list:
kwargs[attribute] = element.get(attribute)
vm = self.domains[int(kwargs["qid"])]
if element.get("uses_default_netvm") is None:
uses_default_netvm = True
else:
uses_default_netvm = (
True if element.get("uses_default_netvm") == "True" else False)
if not uses_default_netvm:
netvm_qid = element.get("netvm_qid")
if netvm_qid is None or netvm_qid == "none":
vm.netvm = None
else:
vm.netvm = int(netvm_qid)
# TODO: dispvm_netvm
def load(self):
qubes_store_file = open(self._store, 'r')
try:
qubes_store_file.seek(0)
tree = lxml.etree.parse(qubes_store_file)
except (EnvironmentError,
xml.parsers.expat.ExpatError) as err:
self.log.error(err)
return False
self.labels = {
1: qubes.Label(1, '0xcc0000', 'red'),
2: qubes.Label(2, '0xf57900', 'orange'),
3: qubes.Label(3, '0xedd400', 'yellow'),
4: qubes.Label(4, '0x73d216', 'green'),
5: qubes.Label(5, '0x555753', 'gray'),
6: qubes.Label(6, '0x3465a4', 'blue'),
7: qubes.Label(7, '0x75507b', 'purple'),
8: qubes.Label(8, '0x000000', 'black'),
}
self.domains.add(qubes.vm.adminvm.AdminVM(
self, None, qid=0, name='dom0'))
vm_classes = ["TemplateVm", "TemplateHVm",
"AppVm", "HVm", "NetVm", "ProxyVm"]
for (vm_class_name) in vm_classes:
vms_of_class = tree.findall("Qubes" + vm_class_name)
# first non-template based, then template based
sorted_vms_of_class = sorted(vms_of_class,
key=lambda x: str(x.get('template_qid')).lower() != "none")
for element in sorted_vms_of_class:
try:
kwargs = {}
if vm_class_name in ["TemplateVm", "TemplateHVm"]:
vm_class = qubes.vm.templatevm.TemplateVM
elif element.get('template_qid').lower() == "none":
kwargs['dir_path'] = element.get('dir_path')
vm_class = StandaloneVM
else:
kwargs['dir_path'] = element.get('dir_path')
kwargs['template'] = int(element.get('template_qid'))
vm_class = AppVM
# simple attributes
for attr in ['installed_by_rpm', 'include_in_backups',
'qrexec_timeout', 'internal', 'label', 'name',
'vcpus', 'memory', 'maxmem', 'default_user',
'debug', 'pci_strictreset', 'mac', 'autostart']:
value = element.get(attr)
if value:
kwargs[attr] = value
# attributes with default value
for attr in ["kernel", "kernelopts"]:
value = element.get(attr)
if value and value.lower() == "none":
value = None
value_is_default = element.get(
"uses_default_{}".format(attr))
if value_is_default and value_is_default.lower() != \
"true":
kwargs[attr] = value
kwargs['hvm'] = "HVm" in vm_class_name
vm = self.add_new_vm(vm_class,
qid=int(element.get('qid')), **kwargs)
services = element.get('services')
if services:
services = eval(services)
else:
services = {}
for service, value in services.iteritems():
feature = service
for repl_feature, repl_service in \
qubes.ext.r3compatibility.\
R3Compatibility.features_to_services.\
iteritems():
if repl_service == service:
feature = repl_feature
vm.features[feature] = value
for attr in ['backup_content', 'backup_path',
'backup_size']:
value = element.get(attr)
vm.features[attr.replace('_', '-')] = value
pcidevs = element.get('pcidevs')
if pcidevs:
pcidevs = eval(pcidevs)
for pcidev in pcidevs:
try:
vm.devices["pci"].attach(pcidev)
except qubes.exc.QubesException as e:
self.log.error("VM {}: {}".format(vm.name, str(e)))
except (ValueError, LookupError) as err:
self.log.error("import error ({1}): {2}".format(
vm_class_name, err))
if 'vm' in locals():
del self.domains[vm]
# After importing all VMs, set netvm references, in the same order
for vm_class_name in vm_classes:
for element in tree.findall("Qubes" + vm_class_name):
try:
self.set_netvm_dependency(element)
except (ValueError, LookupError) as err:
self.log.error("VM {}: failed to set netvm dependency: {}".
format(element.get('name'), err))
self.load_globals(tree.getroot())
def save(self):
raise NotImplementedError("Saving old qubes.xml not supported")

119
qubes/tests/__init__.py
View File

@ -40,6 +40,9 @@ import time
import qubes.config
import qubes.events
import qubes.backup
import qubes.exc
import qubes.vm.standalonevm
XMLPATH = '/var/lib/qubes/qubes-test.xml'
CLASS_XMLPATH = '/var/lib/qubes/qubes-class-test.xml'
@ -482,7 +485,7 @@ class SystemTestsMixin(object):
except (AttributeError, libvirt.libvirtError):
pass
del app.domains[vm]
del app.domains[vm.qid]
del vm
app.save()
@ -628,6 +631,14 @@ class SystemTestsMixin(object):
# noinspection PyAttributeOutsideInit
class BackupTestsMixin(SystemTestsMixin):
class BackupErrorHandler(logging.Handler):
def __init__(self, errors_queue, level=logging.NOTSET):
super(BackupTestsMixin.BackupErrorHandler, self).__init__(level)
self.errors_queue = errors_queue
def emit(self, record):
self.errors_queue.put(record.getMessage())
def setUp(self):
super(BackupTestsMixin, self).setUp()
self.init_default_template()
@ -642,22 +653,17 @@ class BackupTestsMixin(SystemTestsMixin):
shutil.rmtree(self.backupdir)
os.mkdir(self.backupdir)
self.error_handler = self.BackupErrorHandler(self.error_detected,
level=logging.WARNING)
backup_log = logging.getLogger('qubes.backup')
backup_log.addHandler(self.error_handler)
def tearDown(self):
super(BackupTestsMixin, self).tearDown()
shutil.rmtree(self.backupdir)
def print_progress(self, progress):
if self.verbose:
print >> sys.stderr, "\r-> Backing up files: {0}%...".format(progress)
def error_callback(self, message):
self.error_detected.put(message)
if self.verbose:
print >> sys.stderr, "ERROR: {0}".format(message)
def print_callback(self, msg):
if self.verbose:
print msg
backup_log = logging.getLogger('qubes.backup')
backup_log.removeHandler(self.error_handler)
def fill_image(self, path, size=None, sparse=False):
block_size = 4096
@ -686,8 +692,8 @@ class BackupTestsMixin(SystemTestsMixin):
if self.verbose:
print >>sys.stderr, "-> Creating %s" % vmname
testnet = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=vmname, template=template, provides_network=True)
testnet.create_on_disk(verbose=self.verbose)
name=vmname, template=template, provides_network=True, label='red')
testnet.create_on_disk()
vms.append(testnet)
self.fill_image(testnet.private_img, 20*1024*1024)
@ -695,55 +701,68 @@ class BackupTestsMixin(SystemTestsMixin):
if self.verbose:
print >>sys.stderr, "-> Creating %s" % vmname
testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=vmname, template=template)
name=vmname, template=template, label='red')
testvm1.uses_default_netvm = False
testvm1.netvm = testnet
testvm1.create_on_disk(verbose=self.verbose)
testvm1.create_on_disk()
vms.append(testvm1)
self.fill_image(testvm1.private_img, 100*1024*1024)
vmname = self.make_vm_name('testhvm1')
if self.verbose:
print >>sys.stderr, "-> Creating %s" % vmname
testvm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=vmname,
hvm=True)
testvm2.create_on_disk(verbose=self.verbose)
testvm2 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
name=vmname,
hvm=True, label='red')
testvm2.create_on_disk()
self.fill_image(testvm2.root_img, 1024*1024*1024, True)
vms.append(testvm2)
vmname = self.make_vm_name('template')
if self.verbose:
print >>sys.stderr, "-> Creating %s" % vmname
testvm3 = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM,
name=vmname, label='red')
testvm3.create_on_disk()
self.fill_image(testvm3.root_img, 100*1024*1024, True)
vms.append(testvm3)
vmname = self.make_vm_name('custom')
if self.verbose:
print >>sys.stderr, "-> Creating %s" % vmname
testvm4 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=vmname, template=testvm3, label='red')
testvm4.create_on_disk()
vms.append(testvm4)
self.app.save()
return vms
def make_backup(self, vms, prepare_kwargs=dict(), do_kwargs=dict(),
target=None, expect_failure=False):
# XXX: bakup_prepare and backup_do don't support host_collection
# self.qc.unlock_db()
def make_backup(self, vms, target=None, expect_failure=False, **kwargs):
if target is None:
target = self.backupdir
try:
files_to_backup = \
qubes.backup.backup_prepare(vms,
print_callback=self.print_callback,
**prepare_kwargs)
except qubes.qubes.QubesException as e:
backup = qubes.backup.Backup(self.app, vms, **kwargs)
except qubes.exc.QubesException as e:
if not expect_failure:
self.fail("QubesException during backup_prepare: %s" % str(e))
else:
raise
backup.passphrase = 'qubes'
backup.target_dir = target
try:
qubes.backup.backup_do(target, files_to_backup, "qubes",
progress_callback=self.print_progress,
**do_kwargs)
except qubes.qubes.QubesException as e:
backup.backup_do()
except qubes.exc.QubesException as e:
if not expect_failure:
self.fail("QubesException during backup_do: %s" % str(e))
else:
raise
# FIXME why?
self.reload_db()
#self.reload_db()
def restore_backup(self, source=None, appvm=None, options=None,
expect_errors=None):
@ -753,23 +772,18 @@ class BackupTestsMixin(SystemTestsMixin):
else:
backupfile = source
with self.assertNotRaises(qubes.qubes.QubesException):
backup_info = qubes.backup.backup_restore_prepare(
backupfile, "qubes",
host_collection=self.app,
print_callback=self.print_callback,
appvm=appvm,
options=options or {})
with self.assertNotRaises(qubes.exc.QubesException):
restore_op = qubes.backup.BackupRestore(
self.app, backupfile, appvm, "qubes")
if options:
for key, value in options.iteritems():
setattr(restore_op.options, key, value)
restore_info = restore_op.get_restore_info()
if self.verbose:
qubes.backup.backup_restore_print_summary(backup_info)
print restore_op.get_restore_summary(restore_info)
with self.assertNotRaises(qubes.qubes.QubesException):
qubes.backup.backup_restore_do(
backup_info,
host_collection=self.app,
print_callback=self.print_callback if self.verbose else None,
error_callback=self.error_callback)
with self.assertNotRaises(qubes.exc.QubesException):
restore_op.restore_do(restore_info)
# maybe someone forgot to call .save()
self.reload_db()
@ -777,6 +791,9 @@ class BackupTestsMixin(SystemTestsMixin):
errors = []
if expect_errors is None:
expect_errors = []
else:
self.assertFalse(self.error_detected.empty(),
"Restore errors expected, but none detected")
while not self.error_detected.empty():
current_error = self.error_detected.get()
if any(map(current_error.startswith, expect_errors)):
@ -821,8 +838,8 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
'qubes.tests.int.dom0_update',
'qubes.tests.int.network',
# 'qubes.tests.vm_qrexec_gui',
# 'qubes.tests.backup',
# 'qubes.tests.backupcompatibility',
'qubes.tests.int.backup',
'qubes.tests.int.backupcompatibility',
# 'qubes.tests.regressions',
# tool tests

132
tests/backup.py → qubes/tests/int/backup.py
View File

@ -28,40 +28,44 @@ import os
import unittest
import sys
from qubes.qubes import QubesException, QubesTemplateVm
import qubes
import qubes.exc
import qubes.tests
import qubes.vm.appvm
import qubes.vm.templatevm
class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def test_000_basic_backup(self):
vms = self.create_backup_vms()
self.make_backup(vms)
self.remove_vms(vms)
self.remove_vms(reversed(vms))
self.restore_backup()
self.remove_vms(vms)
for vm in vms:
self.assertIn(vm.name, self.app.domains)
def test_001_compressed_backup(self):
vms = self.create_backup_vms()
self.make_backup(vms, do_kwargs={'compressed': True})
self.remove_vms(vms)
self.make_backup(vms, compressed=True)
self.remove_vms(reversed(vms))
self.restore_backup()
self.remove_vms(vms)
for vm in vms:
self.assertIn(vm.name, self.app.domains)
def test_002_encrypted_backup(self):
vms = self.create_backup_vms()
self.make_backup(vms, do_kwargs={'encrypted': True})
self.remove_vms(vms)
self.make_backup(vms, encrypted=True)
self.remove_vms(reversed(vms))
self.restore_backup()
self.remove_vms(vms)
for vm in vms:
self.assertIn(vm.name, self.app.domains)
def test_003_compressed_encrypted_backup(self):
vms = self.create_backup_vms()
self.make_backup(vms,
do_kwargs={
'compressed': True,
'encrypted': True})
self.remove_vms(vms)
self.make_backup(vms, compressed=True, encrypted=True)
self.remove_vms(reversed(vms))
self.restore_backup()
self.remove_vms(vms)
for vm in vms:
self.assertIn(vm.name, self.app.domains)
def test_004_sparse_multipart(self):
vms = []
@ -70,29 +74,36 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
if self.verbose:
print >>sys.stderr, "-> Creating %s" % vmname
hvmtemplate = self.qc.add_new_vm("QubesTemplateHVm", name=vmname)
hvmtemplate.create_on_disk(verbose=self.verbose)
hvmtemplate = self.app.add_new_vm(
qubes.vm.templatevm.TemplateVM, name=vmname, hvm=True, label='red')
hvmtemplate.create_on_disk()
self.fill_image(os.path.join(hvmtemplate.dir_path, '00file'),
195*1024*1024-4096*3)
self.fill_image(hvmtemplate.private_img, 195*1024*1024-4096*3)
self.fill_image(hvmtemplate.root_img, 1024*1024*1024, sparse=True)
vms.append(hvmtemplate)
self.qc.save()
self.app.save()
self.make_backup(vms)
self.remove_vms(vms)
self.remove_vms(reversed(vms))
self.restore_backup()
self.remove_vms(vms)
for vm in vms:
self.assertIn(vm.name, self.app.domains)
# TODO check vm.backup_timestamp
def test_005_compressed_custom(self):
vms = self.create_backup_vms()
self.make_backup(vms, do_kwargs={'compressed': "bzip2"})
self.remove_vms(vms)
self.make_backup(vms, compressed="bzip2")
self.remove_vms(reversed(vms))
self.restore_backup()
self.remove_vms(vms)
for vm in vms:
self.assertIn(vm.name, self.app.domains)
def test_100_backup_dom0_no_restore(self):
self.make_backup([self.qc[0]])
# do not write it into dom0 home itself...
os.mkdir('/var/tmp/test-backup')
self.backupdir = '/var/tmp/test-backup'
self.make_backup([self.app.domains[0]])
# TODO: think of some safe way to test restore...
def test_200_restore_over_existing_directory(self):
@ -102,7 +113,7 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
"""
vms = self.create_backup_vms()
self.make_backup(vms)
self.remove_vms(vms)
self.remove_vms(reversed(vms))
test_dir = vms[0].dir_path
os.mkdir(test_dir)
with open(os.path.join(test_dir, 'some-file.txt'), 'w') as f:
@ -112,7 +123,6 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
'*** Directory {} already exists! It has been moved'.format(
test_dir)
])
self.remove_vms(vms)
def test_210_auto_rename(self):
"""
@ -122,58 +132,51 @@ class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
vms = self.create_backup_vms()
self.make_backup(vms)
self.restore_backup(options={
'rename-conflicting': True
'rename_conflicting': True
})
for vm in vms:
self.assertIsNotNone(self.qc.get_vm_by_name(vm.name+'1'))
restored_vm = self.qc.get_vm_by_name(vm.name+'1')
if vm.netvm and not vm.uses_default_netvm:
self.assertEqual(restored_vm.netvm.name, vm.netvm.name+'1')
with self.assertNotRaises(
(qubes.exc.QubesVMNotFoundError, KeyError)):
restored_vm = self.app.domains[vm.name + '1']
if vm.netvm and not vm.property_is_default('netvm'):
self.assertEqual(restored_vm.netvm.name, vm.netvm.name + '1')
self.remove_vms(vms)
class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin):
def setUp(self):
super(TC_10_BackupVMMixin, self).setUp()
self.backupvm = self.qc.add_new_vm(
"QubesAppVm",
self.backupvm = self.app.add_new_vm(
qubes.vm.appvm.AppVM,
label='red',
name=self.make_vm_name('backupvm'),
template=self.qc.get_vm_by_name(self.template)
template=self.template
)
self.backupvm.create_on_disk(verbose=self.verbose)
self.backupvm.create_on_disk()
def test_100_send_to_vm_file_with_spaces(self):
vms = self.create_backup_vms()
self.backupvm.start()
self.backupvm.run("mkdir '/var/tmp/backup directory'", wait=True)
self.make_backup(vms,
do_kwargs={
'appvm': self.backupvm,
'compressed': True,
'encrypted': True},
target='/var/tmp/backup directory')
self.remove_vms(vms)
self.make_backup(vms, target_vm=self.backupvm,
compressed=True, encrypted=True,
target='/var/tmp/backup directory')
self.remove_vms(reversed(vms))
p = self.backupvm.run("ls /var/tmp/backup*/qubes-backup*",
passio_popen=True)
(backup_path, _) = p.communicate()
backup_path = backup_path.strip()
self.restore_backup(source=backup_path,
appvm=self.backupvm)
self.remove_vms(vms)
def test_110_send_to_vm_command(self):
vms = self.create_backup_vms()
self.backupvm.start()
self.make_backup(vms,
do_kwargs={
'appvm': self.backupvm,
'compressed': True,
'encrypted': True},
target='dd of=/var/tmp/backup-test')
self.remove_vms(vms)
self.make_backup(vms, target_vm=self.backupvm,
compressed=True, encrypted=True,
target='dd of=/var/tmp/backup-test')
self.remove_vms(reversed(vms))
self.restore_backup(source='dd if=/var/tmp/backup-test',
appvm=self.backupvm)
self.remove_vms(vms)
def test_110_send_to_vm_no_space(self):
"""
@ -192,27 +195,18 @@ class TC_10_BackupVMMixin(qubes.tests.BackupTestsMixin):
user="root", wait=True)
if retcode != 0:
raise RuntimeError("Failed to prepare backup directory")
with self.assertRaises(QubesException):
self.make_backup(vms,
do_kwargs={
'appvm': self.backupvm,
'compressed': False,
'encrypted': True},
target='/home/user/backup',
expect_failure=True)
self.qc.lock_db_for_writing()
self.qc.load()
self.remove_vms(vms)
with self.assertRaises(qubes.exc.QubesException):
self.make_backup(vms, target_vm=self.backupvm,
compressed=False, encrypted=True,
target='/home/user/backup',
expect_failure=True)
def load_tests(loader, tests, pattern):
try:
qc = qubes.qubes.QubesVmCollection()
qc.lock_db_for_reading()
qc.load()
qc.unlock_db()
templates = [vm.name for vm in qc.values() if
isinstance(vm, QubesTemplateVm)]
app = qubes.Qubes()
templates = [vm.name for vm in app.domains if
isinstance(vm, qubes.vm.templatevm.TemplateVM)]
except OSError:
templates = []
for template in templates:

88
tests/backupcompatibility.py → qubes/tests/int/backupcompatibility.py
View File

@ -29,8 +29,6 @@ import subprocess
import unittest
import sys
import re
from qubes.qubes import QubesVmCollection, QubesException
from qubes import backup
import qubes.tests
@ -146,6 +144,11 @@ compression-filter=gzip
'''
class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def tearDown(self):
self.remove_test_vms(prefix="test-")
super(TC_00_BackupCompatibility, self).tearDown()
def create_whitelisted_appmenus(self, filename):
f = open(filename, "w")
f.write("gnome-terminal.desktop\n")
@ -401,19 +404,22 @@ class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, qubes.tests.QubesT
f.write(QUBESXML_R1)
f.close()
self.restore_backup(self.backupdir, options={
'use-default-template': True,
'use-default-netvm': True,
})
self.assertIsNotNone(self.qc.get_vm_by_name("test-template-clone"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testproxy"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-work"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-standalonevm"))
self.assertIsNotNone(self.qc.get_vm_by_name(
"test-custom-template-appvm"))
self.assertEqual(self.qc.get_vm_by_name("test-custom-template-appvm")
self.restore_backup(self.backupdir,
options={
'use-default-template': True,
'use-default-netvm': True,
},
expect_errors=['Kernel None not installed, using default one']
)
with self.assertNotRaises(KeyError):
vm = self.app.domains["test-template-clone"]
vm = self.app.domains["test-testproxy"]
vm = self.app.domains["test-work"]
vm = self.app.domains["test-standalonevm"]
vm = self.app.domains["test-custom-template-appvm"]
self.assertEqual(self.app.domains["test-custom-template-appvm"]
.template,
self.qc.get_vm_by_name("test-template-clone"))
self.app.domains["test-template-clone"])
def test_200_r2b2(self):
self.create_v1_files(r2b2=True)
@ -425,16 +431,16 @@ class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, qubes.tests.QubesT
self.restore_backup(self.backupdir, options={
'use-default-template': True,
})
self.assertIsNotNone(self.qc.get_vm_by_name("test-template-clone"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testproxy"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-work"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testhvm"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-standalonevm"))
self.assertIsNotNone(self.qc.get_vm_by_name(
"test-custom-template-appvm"))
self.assertEqual(self.qc.get_vm_by_name("test-custom-template-appvm")
with self.assertNotRaises(KeyError):
vm = self.app.domains["test-template-clone"]
vm = self.app.domains["test-testproxy"]
vm = self.app.domains["test-work"]
vm = self.app.domains["test-testhvm"]
vm = self.app.domains["test-standalonevm"]
vm = self.app.domains["test-custom-template-appvm"]
self.assertEqual(self.app.domains["test-custom-template-appvm"]
.template,
self.qc.get_vm_by_name("test-template-clone"))
self.app.domains["test-template-clone"])
def test_210_r2(self):
self.create_v3_backup(False)
@ -443,16 +449,16 @@ class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, qubes.tests.QubesT
'use-default-template': True,
'use-default-netvm': True,
})
self.assertIsNotNone(self.qc.get_vm_by_name("test-template-clone"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testproxy"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-work"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testhvm"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-standalonevm"))
self.assertIsNotNone(self.qc.get_vm_by_name(
"test-custom-template-appvm"))
self.assertEqual(self.qc.get_vm_by_name("test-custom-template-appvm")
with self.assertNotRaises(KeyError):
vm = self.app.domains["test-template-clone"]
vm = self.app.domains["test-testproxy"]
vm = self.app.domains["test-work"]
vm = self.app.domains["test-testhvm"]
vm = self.app.domains["test-standalonevm"]
vm = self.app.domains["test-custom-template-appvm"]
self.assertEqual(self.app.domains["test-custom-template-appvm"]
.template,
self.qc.get_vm_by_name("test-template-clone"))
self.app.domains["test-template-clone"])
def test_220_r2_encrypted(self):
self.create_v3_backup(True)
@ -461,13 +467,13 @@ class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, qubes.tests.QubesT
'use-default-template': True,
'use-default-netvm': True,
})
self.assertIsNotNone(self.qc.get_vm_by_name("test-template-clone"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testproxy"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-work"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-testhvm"))
self.assertIsNotNone(self.qc.get_vm_by_name("test-standalonevm"))
self.assertIsNotNone(self.qc.get_vm_by_name(
"test-custom-template-appvm"))
self.assertEqual(self.qc.get_vm_by_name("test-custom-template-appvm")
with self.assertNotRaises(KeyError):
vm = self.app.domains["test-template-clone"]
vm = self.app.domains["test-testproxy"]
vm = self.app.domains["test-work"]
vm = self.app.domains["test-testhvm"]
vm = self.app.domains["test-standalonevm"]
vm = self.app.domains["test-custom-template-appvm"]
self.assertEqual(self.app.domains["test-custom-template-appvm"]
.template,
self.qc.get_vm_by_name("test-template-clone"))
self.app.domains["test-template-clone"])

16
qubes/tests/vm/qubesvm.py
View File

@ -639,22 +639,6 @@ class TC_90_QubesVM(qubes.tests.QubesTestCase):
vm = self.get_vm()
self._test_generic_bool_property(vm, 'pci_strictreset')
def test_380_backup_size(self):
vm = self.get_vm()
self.assertPropertyDefaultValue(vm, 'backup_size', 0)
self.assertPropertyValue(vm, 'backup_size', 0, 0, '0')
del vm.backup_size
self.assertPropertyDefaultValue(vm, 'backup_size', 0)
self.assertPropertyValue(vm, 'backup_size', '0', 0, '0')
self.assertPropertyValue(vm, 'backup_size', 300, 300, '300')
def test_390_backup_path(self):
vm = self.get_vm()
self.assertPropertyDefaultValue(vm, 'backup_path', '')
self.assertPropertyValue(vm, 'backup_path', 'some/dir', 'some/dir')
del vm.backup_path
self.assertPropertyDefaultValue(vm, 'backup_path', '')
def test_400_backup_timestamp(self):
vm = self.get_vm()
timestamp = datetime.datetime(2016, 1, 1, 12, 14, 2)

44
qubes/utils.py
View File

@ -90,9 +90,12 @@ def format_doc(docstring):
# maybe adapt https://code.activestate.com/recipes/578019
def parse_size(size):
units = [
('K', 1024), ('KB', 1024),
('M', 1024*1024), ('MB', 1024*1024),
('G', 1024*1024*1024), ('GB', 1024*1024*1024),
('K', 1000), ('KB', 1000),
('M', 1000 * 1000), ('MB', 1000 * 1000),
('G', 1000 * 1000 * 1000), ('GB', 1000 * 1000 * 1000),
('Ki', 1024), ('KiB', 1024),
('Mi', 1024 * 1024), ('MiB', 1024 * 1024),
('Gi', 1024 * 1024 * 1024), ('GiB', 1024 * 1024 * 1024),
]
size = size.strip().upper()
@ -102,10 +105,43 @@ def parse_size(size):
for unit, multiplier in units:
if size.endswith(unit):
size = size[:-len(unit)].strip()
return int(size)*multiplier
return int(size) * multiplier
raise qubes.exc.QubesException("Invalid size: {0}.".format(size))
def mbytes_to_kmg(size):
if size > 1024:
return "%d GiB" % (size / 1024)
else:
return "%d MiB" % size
def kbytes_to_kmg(size):
if size > 1024:
return mbytes_to_kmg(size / 1024)
else:
return "%d KiB" % size
def bytes_to_kmg(size):
if size > 1024:
return kbytes_to_kmg(size / 1024)
else:
return "%d B" % size
def size_to_human(size):
"""Humane readable size, with 1/10 precision"""
if size < 1024:
return str(size)
elif size < 1024 * 1024:
return str(round(size / 1024.0, 1)) + ' KiB'
elif size < 1024 * 1024 * 1024:
return str(round(size / (1024.0 * 1024), 1)) + ' MiB'
else:
return str(round(size / (1024.0 * 1024 * 1024), 1)) + ' GiB'
def urandom(size):
rand = os.urandom(size)
if rand is None:

15
qubes/vm/qubesvm.py
View File

@ -262,22 +262,11 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
doc='''Setting this to `True` means that VM should be autostarted on
dom0 boot.''')
# XXX I don't understand backups
include_in_backups = qubes.property('include_in_backups', default=True,
include_in_backups = qubes.property('include_in_backups',
default=(lambda self: not self.internal),
type=bool, setter=qubes.property.bool,
doc='If this domain is to be included in default backup.')
backup_content = qubes.property('backup_content', default=False,
type=bool, setter=qubes.property.bool,
doc='FIXME')
backup_size = qubes.property('backup_size', type=int, default=0,
doc='FIXME')
# TODO default=None?
backup_path = qubes.property('backup_path', type=str, default='',
doc='FIXME')
# format got changed from %s to str(datetime.datetime)
backup_timestamp = qubes.property('backup_timestamp', default=None,
setter=(lambda self, prop, value:

4
rpm_spec/core-dom0.spec
View File

@ -201,7 +201,9 @@ fi
%dir %{python_sitelib}/qubes
%{python_sitelib}/qubes/__init__.py*
%{python_sitelib}/qubes/backup.py*
%{python_sitelib}/qubes/config.py*
%{python_sitelib}/qubes/core2migration.py*
%{python_sitelib}/qubes/devices.py*
%{python_sitelib}/qubes/dochelpers.py*
%{python_sitelib}/qubes/events.py*
@ -272,6 +274,8 @@ fi
%dir %{python_sitelib}/qubes/tests/int
%{python_sitelib}/qubes/tests/int/__init__.py*
%{python_sitelib}/qubes/tests/int/backup.py*
%{python_sitelib}/qubes/tests/int/backupcompatibility.py*
%{python_sitelib}/qubes/tests/int/basic.py*
%{python_sitelib}/qubes/tests/int/dom0_update.py*
%{python_sitelib}/qubes/tests/int/network.py*