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

qubes: pylint fixes (minor programming errors)

Browse Source
This commit is contained in:
Wojtek Porczyk 2015-01-22 11:24:23 +01:00
parent 186b277418
commit 5d9b92a039
5 changed files with 37 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
qubes/__init__.py
View File

@ -176,6 +176,8 @@ class QubesHost(object):
def __init__(self, app):
self.app = app
self._no_cpus = None
self._total_mem = None
self._physinfo = None
def _fetch(self):
@ -256,7 +258,7 @@ class QubesHost(object):
current_time = time.time()
current = {}
try:
info = self._app.vmm.xc.domain_getinfo(0, qubes.config.max_qid)
info = self.app.vmm.xc.domain_getinfo(0, qubes.config.max_qid)
except AttributeError:
raise NotImplementedError(
'This function requires Xen hypervisor')
@ -433,7 +435,7 @@ class VMCollection(object):
self.__class__.__name__))
if not hasattr(value, 'qid'):
value.qid = self.domains.get_new_unused_qid()
value.qid = self.get_new_unused_qid()
if value.qid in self:
raise ValueError('This collection already holds VM that has '
@ -499,7 +501,7 @@ class VMCollection(object):
continue
dependent_vms.add(vm.qid)
# if vm.is_netvm():
new_vms.append(vm.qid)
new_vms.add(vm.qid)
return dependent_vms
@ -994,7 +996,8 @@ class PropertyHolder(qubes.events.Emitter):
if hard:
raise AssertionError(msg)
else:
self.log(msg)
# pylint: disable=no-member
self.log.fatal(msg)
import qubes.vm
@ -1030,7 +1033,7 @@ class VMProperty(property):
def __set__(self, instance, value):
if value is None:
if self.allow_none:
super(VMProperty, self).__set__(self, instance, vm)
super(VMProperty, self).__set__(self, instance, value)
return
else:
raise ValueError(
@ -1143,6 +1146,7 @@ class Qubes(PropertyHolder):
self.host = QubesHost(self)
self._store = store
self._storefd = None
self.load()
@ -1154,7 +1158,7 @@ class Qubes(PropertyHolder):
:raises OSError: on failure
:raises lxml.etree.XMLSyntaxError: on syntax error in qubes.xml
'''
if hasattr(self, '_storefd'):
if self._storefd is not None:
return
try:
@ -1198,12 +1202,12 @@ class Qubes(PropertyHolder):
return
# stage 1: load labels
for node in self._xml.xpath('./labels/label'):
for node in self.xml.xpath('./labels/label'):
label = Label.fromxml(node)
self.labels[label.id] = label
self.labels[label.index] = label
# stage 2: load VMs
for node in self._xml.xpath('./domains/domain'):
for node in self.xml.xpath('./domains/domain'):
# pylint: disable=no-member
cls = qubes.vm.BaseVM.register[node.get('class')]
vm = cls(self, node)
@ -1266,9 +1270,9 @@ class Qubes(PropertyHolder):
def __del__(self):
# intentionally do not call explicit unlock to not unlock the file
# before all buffers are flushed
if hasattr(self, '_storefd'):
if self._storefd is not None:
self._storefd.close()
del self._storefd
self._storefd = None
def __xml__(self):
@ -1321,7 +1325,7 @@ class Qubes(PropertyHolder):
def on_domain_pre_deleted(self, event, vm):
# pylint: disable=unused-argument
if isinstance(vm, qubes.vm.templatevm.TemplateVM):
appvms = self.get_vms_based_on(vm)
appvms = self.domains.get_vms_based_on(vm)
if appvms:
raise QubesException(
'Cannot remove template that has dependent AppVMs. '

8
qubes/storage/__init__.py
View File

@ -114,7 +114,7 @@ class VMStorage(object):
source, destination))
def get_disk_utilization(self):
return qubes.utils.get_disk_usage(self.vmdir)
return qubes.utils.get_disk_usage(self.vm.dir_path)
def get_disk_utilization_private_img(self):
# pylint: disable=invalid-name
@ -142,7 +142,7 @@ class VMStorage(object):
old_umask = os.umask(002)
self.vm.log.info('Creating directory: {0}'.format(self.vm.dir_path))
os.mkdir(self.vmdir)
os.mkdir(self.vm.dir_path)
self.create_on_disk_private_img(source_template)
self.create_on_disk_root_img(source_template)
self.reset_volatile_storage(source_template)
@ -188,7 +188,7 @@ class VMStorage(object):
def verify_files(self):
if not os.path.exists(self.vm.dir_path):
raise qubes.QubesException(
'VM directory does not exist: {}'.format(self.vmdir))
'VM directory does not exist: {}'.format(self.vm.dir_path))
if hasattr(self.vm, 'root_img') and not os.path.exists(self.root_img):
raise qubes.QubesException(
@ -239,7 +239,7 @@ class VMStorage(object):
and not os.path.exists(self.private_img):
self.vm.log.info('Creating empty VM private image file: {0}'.format(
self.private_img))
self.storage.create_on_disk_private_img()
self.create_on_disk_private_img()
def get_storage(vm):

2
qubes/storage/xen.py
View File

@ -79,7 +79,7 @@ class XenVMStorage(qubes.storage.VMStorage):
def _get_rootdev(self):
if isinstance(self.vm, qubes.vm.templatevm.TemplateVM):
return self._format_disk_dev(
'{}:{}'.format(self.root_img, self.rootcow_img),
'{}:{}'.format(self.root_img, self.vm.rootcow_img),
self.root_dev,
script='block-origin')

4
qubes/tests/init.py
View File

@ -145,7 +145,7 @@ class TC_10_property(qubes.tests.QubesTestCase):
del self.holder.testprop1
with self.assertRaises(AttributeError):
self.holder.testprop
self.holder.testprop1
def test_090_delete_by_assign(self):
self.holder.testprop1 = 'testvalue'
@ -158,7 +158,7 @@ class TC_10_property(qubes.tests.QubesTestCase):
self.holder.testprop1 = qubes.property.DEFAULT
with self.assertRaises(AttributeError):
self.holder.testprop
self.holder.testprop1
def test_092_delete_default(self):
class MyTestHolder(qubes.tests.TestEmitter, qubes.PropertyHolder):

28
qubes/vm/qubesvm.py
View File

@ -351,7 +351,7 @@ class QubesVM(qubes.vm.BaseVM):
@property
def updateable(self):
'''True if this machine may be updated on its own.'''
return hasattr(self, 'template')
return not hasattr(self, 'template')
@property
@ -521,16 +521,16 @@ class QubesVM(qubes.vm.BaseVM):
# self.netvm.post_vm_net_detach(self)
if new_netvm is None:
if not self._do_not_reset_firewall:
# Set also firewall to block all traffic as discussed in #370
if os.path.exists(self.firewall_conf):
shutil.copy(self.firewall_conf,
os.path.join(qubes.config.system_path['qubes_base_dir'],
'backup',
'%s-firewall-%s.xml' % (self.name,
time.strftime('%Y-%m-%d-%H:%M:%S'))))
self.write_firewall_conf({'allow': False, 'allowDns': False,
'allowIcmp': False, 'allowYumProxy': False, 'rules': []})
# if not self._do_not_reset_firewall:
# Set also firewall to block all traffic as discussed in #370
if os.path.exists(self.firewall_conf):
shutil.copy(self.firewall_conf,
os.path.join(qubes.config.system_path['qubes_base_dir'],
'backup',
'%s-firewall-%s.xml' % (self.name,
time.strftime('%Y-%m-%d-%H:%M:%S'))))
self.write_firewall_conf({'allow': False, 'allowDns': False,
'allowIcmp': False, 'allowYumProxy': False, 'rules': []})
else:
new_netvm.connected_vms.add(self)
@ -744,9 +744,9 @@ class QubesVM(qubes.vm.BaseVM):
if qmemman_present:
qmemman_client.close()
if self._start_guid_first and start_guid and not preparing_dvm \
and os.path.exists('/var/run/shm.id'):
self.start_guid()
# if self._start_guid_first and start_guid and not preparing_dvm \
# and os.path.exists('/var/run/shm.id'):
# self.start_guid()
if not preparing_dvm:
self.start_qrexec_daemon()