qubes: pylint fixes

Fix bunch of errors and warnings.
This commit is contained in:
Wojtek Porczyk 2015-10-05 23:46:25 +02:00
parent 15713cbf46
commit ea44c0acf3
28 changed files with 132 additions and 135 deletions

View File

@ -42,7 +42,8 @@ disable=
duplicate-code,
star-args,
cyclic-import,
abstract-class-little-used
abstract-class-little-used,
bad-continuation
[REPORTS]
@ -128,7 +129,7 @@ variable-rgx=[a-z_][a-z0-9_]{2,30}$
inlinevar-rgx=[A-Za-z_][A-Za-z0-9_]*$
# Good variable names which should always be accepted, separated by a comma
good-names=e,i,j,k,m,p,ex,Run,_,log,vm,xc,xs,ip,fd,rw
good-names=e,i,j,k,m,p,ex,Run,_,log,vm,xc,xs,ip,fd,fh,rw,st
# Bad variable names which should always be refused, separated by a comma
bad-names=foo,bar,baz,toto,tutu,tata

View File

@ -43,8 +43,6 @@ from qubes.qdb import QubesDB,Error,DisconnectedError
import xen.lowlevel.xc
import xen.lowlevel.xs
BLKSIZE = 512
# all frontends, prefer xvdi
# TODO: get this from libvirt driver?
AVAILABLE_FRONTENDS = ['xvd'+c for c in
@ -79,27 +77,6 @@ def size_to_human (size):
else:
return str(round(size/(1024.0*1024*1024),1)) + ' GiB'
def get_disk_usage_one(st):
try:
return st.st_blocks * BLKSIZE
except AttributeError:
return st.st_size
def get_disk_usage(path):
try:
st = os.lstat(path)
except OSError:
return 0
ret = get_disk_usage_one(st)
# if path is not a directory, this is skipped
for dirpath, dirnames, filenames in os.walk(path):
for name in dirnames + filenames:
ret += get_disk_usage_one(os.lstat(os.path.join(dirpath, name)))
return ret
def print_stdout(text):
print (text)

View File

@ -1278,6 +1278,7 @@ class Qubes(PropertyHolder):
if os.name == 'posix':
fcntl.lockf(fh, fcntl.LOCK_EX)
elif os.name == 'nt':
# pylint: disable=protected-access
overlapped = pywintypes.OVERLAPPED()
win32file.LockFileEx(
win32file._get_osfhandle(fh.fileno()),
@ -1351,9 +1352,9 @@ class Qubes(PropertyHolder):
pass
# then search for name
for l in self.labels.values():
if l.name == label:
return l
for i in self.labels.values():
if i.name == label:
return i
# last call, if label is a number represented as str, search in indices
try:

View File

@ -22,6 +22,8 @@
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''Constants which can be configured in one place'''
qubes_base_dir = "/var/lib/qubes"
system_path = {
'qubes_guid_path': '/usr/bin/qubes-guid',

View File

@ -48,6 +48,7 @@ def handler(*events):
'''
def decorator(func):
# pylint: disable=missing-docstring
func.ha_events = events
return func

View File

@ -38,10 +38,12 @@ class QubesManagerExtension(qubes.ext.Extension):
super(QubesManagerExtension, self).__init__(*args, **kwargs)
self._system_bus = dbus.SystemBus()
# pylint: disable=no-self-use,unused-argument,too-few-public-methods
@qubes.ext.handler('status:error')
def on_status_error(self, vm, status, message):
try:
qubes_manager = system_bus.get_object(
qubes_manager = self._system_bus.get_object(
'org.qubesos.QubesManager',
'/org/qubesos/QubesManager')
qubes_manager.notify_error(vm.name, message,
@ -51,9 +53,9 @@ class QubesManagerExtension(qubes.ext.Extension):
pass
@qubes.ext.handler('status:no-error')
def on_status_error(self, vm, status, message):
def on_status_no_error(self, vm, status, message):
try:
qubes_manager = system_bus.get_object(
qubes_manager = self._system_bus.get_object(
'org.qubesos.QubesManager',
'/org/qubesos/QubesManager')
qubes_manager.clear_error_exact(vm.name, message,

View File

@ -61,7 +61,7 @@ class DBusHandler(logging.Handler):
'org.freedesktop.Notifications', '/org/freedesktop/Notifications')
def handle(self, record):
def emit(self, record):
app_icon = self.app_icons[
max(level for level in self.app_icons if level <= record.levelno)]

View File

@ -1,5 +1,7 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,7 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,7 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -37,6 +37,8 @@ import sys
import qubes
import qubes.utils
BLKSIZE = 512
class VMStorage(object):
'''Class for handling VM virtual disks.
@ -90,7 +92,7 @@ class VMStorage(object):
:py:attr:`self.vm.dir_path`
'''
return os.path.join(qubes.config.system_path['qubes_base_dir'],
qubes.config.system_path['qubes_kernels_base_dir'], self.vm.kernel) \
qubes.config.system_path['qubes_kernels_base_dir'], self.vm.kernel)\
if self.vm.kernel is not None \
else os.path.join(self.vm.dir_path,
qubes.config.vm_files['kernels_subdir'])
@ -139,11 +141,11 @@ class VMStorage(object):
source, destination))
def get_disk_utilization(self):
return qubes.utils.get_disk_usage(self.vm.dir_path)
return get_disk_usage(self.vm.dir_path)
def get_disk_utilization_private_img(self):
# pylint: disable=invalid-name
return qubes.utils.get_disk_usage(self.private_img)
return get_disk_usage(self.private_img)
def get_private_img_sz(self):
if not os.path.exists(self.private_img):
@ -267,6 +269,47 @@ class VMStorage(object):
self.create_on_disk_private_img()
def get_disk_usage_one(st):
'''Extract disk usage of one inode from its stat_result struct.
If known, get real disk usage, as written to device by filesystem, not
logical file size. Those values may be different for sparse files.
:param os.stat_result st: stat result
:returns: disk usage
'''
try:
return st.st_blocks * BLKSIZE
except AttributeError:
return st.st_size
def get_disk_usage(path):
'''Get real disk usage of given path (file or directory).
When *path* points to directory, then it is evaluated recursively.
This function tries estiate real disk usage. See documentation of
:py:func:`get_disk_usage_one`.
:param str path: path to evaluate
:returns: disk usage
'''
try:
st = os.lstat(path)
except OSError:
return 0
ret = get_disk_usage_one(st)
# if path is not a directory, this is skipped
for dirpath, dirnames, filenames in os.walk(path):
for name in dirnames + filenames:
ret += get_disk_usage_one(os.lstat(os.path.join(dirpath, name)))
return ret
def get_storage(vm):
'''Factory yielding storage class instances for domains.

View File

@ -45,6 +45,7 @@ class TC_00_Emitter(qubes.tests.QubesTestCase):
def test_001_decorator(self):
class TestEmitter(qubes.events.Emitter):
def __init__(self):
# pylint: disable=bad-super-call
super(TestEmitter, self).__init__()
self.testevent_fired = False

View File

@ -23,7 +23,6 @@
#
import curses
import importlib
import socket
import sys
import unittest

View File

@ -73,6 +73,7 @@ class TC_00_PropertyAction(qubes.tests.QubesTestCase):
args.properties)
def test_003_set_prop_override_default(self):
# pylint: disable=invalid-name
parser = argparse.ArgumentParser()
parser.add_argument('--property', '-p',
action=qubes.tools.PropertyAction)

View File

@ -35,7 +35,7 @@ import qubes.log
class PropertyAction(argparse.Action):
'''Action for argument parser that stores a property.'''
# pylint: disable=redefined-builtin
# pylint: disable=redefined-builtin,too-few-public-methods
def __init__(self,
option_strings,
dest,
@ -49,7 +49,7 @@ class PropertyAction(argparse.Action):
try:
prop, value = values.split('=', 1)
except ValueError:
parser.error('invalid property token: {!r}'.format(token))
parser.error('invalid property token: {!r}'.format(values))
getattr(namespace, self.dest)[prop] = value
@ -57,7 +57,7 @@ class PropertyAction(argparse.Action):
class SinglePropertyAction(argparse.Action):
'''Action for argument parser that stores a property.'''
# pylint: disable=redefined-builtin
# pylint: disable=redefined-builtin,too-few-public-methods
def __init__(self,
option_strings,
dest,

View File

@ -1,5 +1,7 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -26,7 +26,7 @@
# TODO allow to set properties and create domains
import argparse
import sys
import qubes
import qubes.tools
@ -48,7 +48,7 @@ def main(args=None):
'''
args = parser.parse_args(args)
app = qubes.Qubes.create_empty_store(args.app, **args.properties)
qubes.Qubes.create_empty_store(args.app, **args.properties)
return True

View File

@ -71,7 +71,6 @@ parser.add_argument('name', metavar='VMNAME',
nargs='?',
help='name of the domain to create')
#parser.add_option ("-q", "--quiet", action="store_false", dest="verbose", default=True)
def main(args=None):
args = parser.parse_args(args)
@ -83,13 +82,14 @@ def main(args=None):
parser.error('VMNAME is mandatory')
try:
label = args.app.get_label(args.properties['label'])
args.app.get_label(args.properties['label'])
except KeyError:
parser.error('no such label: {!r}; available: {}'.format(args.label,
parser.error('no such label: {!r}; available: {}'.format(
args.properties['label'],
', '.join(repr(l.name) for l in args.app.labels)))
try:
cls = qubes.vm.BaseVM.register[args.cls]
cls = qubes.vm.BaseVM.register[args.cls] # pylint: disable=no-member
except KeyError:
parser.error('no such domain class: {!r}'.format(args.cls))
@ -99,6 +99,8 @@ def main(args=None):
vm = args.app.add_new_vm(cls, **args.properties)
# pylint: disable=line-too-long
# if not options.standalone and any([options.root_copy_from, options.root_move_from]):
# print >> sys.stderr, "root.img can be specified only for standalone VMs"
# exit (1)
@ -132,13 +134,13 @@ def main(args=None):
# if (options.verbose):
# print "--> Replacing root.img with provided file"
os.unlink(vm.root_img)
os.rename(options.root_move_from, vm.root_img)
os.rename(args.root_move_from, vm.root_img)
elif args.root_copy_from is not None:
# if (options.verbose):
# print "--> Replacing root.img with provided file"
os.unlink(vm.root_img)
# use 'cp' to preserve sparse file
subprocess.check_call(['cp', options.root_copy_from, vm.root_img])
subprocess.check_call(['cp', args.root_copy_from, vm.root_img])
except (IOError, OSError) as err:
parser.error(str(err))

View File

@ -25,6 +25,8 @@
'''qvm-kill - forceful shutdown'''
import sys
import qubes
import qubes.tools
parser = qubes.tools.QubesArgumentParser(
@ -43,7 +45,7 @@ def main(args=None):
try:
args.vm.force_shutdown()
except (IOError, OSError, QubesException) as e:
except (IOError, OSError, qubes.QubesException) as e:
parser.error_runtime(str(e))
return True

View File

@ -31,7 +31,6 @@ from __future__ import print_function
import __builtin__
import argparse
import collections
import os
import sys
import textwrap
@ -556,7 +555,7 @@ def get_parser():
wrapper = textwrap.TextWrapper(width=80, break_on_hyphens=False,
initial_indent=' ', subsequent_indent=' ')
parser = qubes.tools.get_parser_base(
parser = qubes.tools.QubesArgumentParser(
formatter_class=argparse.RawTextHelpFormatter,
description='List Qubes domains and their parametres.',
epilog='available formats (see --help-formats):\n{}\n\n'

View File

@ -27,8 +27,6 @@
from __future__ import print_function
import argparse
import os
import subprocess
import sys
import textwrap
@ -40,7 +38,7 @@ import qubes.vm
class _HelpPropertiesAction(argparse.Action):
'''Action for argument parser that displays all properties and exits.'''
# pylint: disable=redefined-builtin
# pylint: disable=redefined-builtin,too-few-public-methods
def __init__(self,
option_strings,
dest=argparse.SUPPRESS,
@ -55,6 +53,7 @@ class _HelpPropertiesAction(argparse.Action):
help=help)
def __call__(self, parser, namespace, values, option_string=None):
# pylint: disable=redefined-outer-name
properties = qubes.vm.qubesvm.QubesVM.property_list()
width = max(len(prop.__name__) for prop in properties)
wrapper = textwrap.TextWrapper(width=80,

View File

@ -27,11 +27,12 @@
# TODO notification in tray
import argparse
import sys
import qubes
class DriveAction(argparse.Action):
'''Action for argument parser that stores drive image path.'''
# pylint: disable=redefined-builtin
# pylint: disable=redefined-builtin,too-few-public-methods
def __init__(self,
option_strings,
dest='drive',
@ -44,7 +45,8 @@ class DriveAction(argparse.Action):
self.prefix = prefix
def __call__(self, parser, namespace, values, option_string=None):
setattr(namespace, self.dest, prefix + value)
# pylint: disable=redefined-outer-name
setattr(namespace, self.dest, self.prefix + values)
parser = qubes.tools.QubesArgumentParser(
@ -85,10 +87,8 @@ parser.add_argument('--preparing-dvm',
parser.add_argument('--no-start-guid',
action='store_false', dest='start_guid', default=True,
help='do actions necessary when preparing DVM image')
help='do not start the gui daemon (ignored)')
#parser.add_option ("--no-guid", action="store_true", dest="noguid", default=False,
# help="Do not start the GUId (ignored)")
#parser.add_option ("--tray", action="store_true", dest="tray", default=False,
# help="Use tray notifications instead of stdout" )
@ -126,7 +126,8 @@ def main(args=None):
'error verifying files for domain {!r}: {!r}'.format(vm.name, e))
try:
xid = vm.start(
#xid =
vm.start(
preparing_dvm=args.preparing_dvm,
start_guid=args.start_guid)
# notify_function=

View File

@ -32,6 +32,8 @@ import docutils
import docutils.core
import docutils.io
import qubes
def get_timezone():
# fc18
@ -86,7 +88,8 @@ def format_doc(docstring):
# FIXME those are wrong, k/M/G are SI prefixes and means 10**3
# maybe adapt https://code.activestate.com/recipes/578019
def parse_size(size):
units = [ ('K', 1024), ('KB', 1024),
units = [
('K', 1024), ('KB', 1024),
('M', 1024*1024), ('MB', 1024*1024),
('G', 1024*1024*1024), ('GB', 1024*1024*1024),
]
@ -100,4 +103,4 @@ def parse_size(size):
size = size[:-len(unit)].strip()
return int(size)*multiplier
raise QubesException("Invalid size: {0}.".format(size))
raise qubes.QubesException("Invalid size: {0}.".format(size))

View File

@ -206,44 +206,6 @@ class BaseVM(qubes.PropertyHolder):
self.log = qubes.log.get_vm_logger(self.name)
def add_new_vm(self, vm):
'''Add new Virtual Machine to colletion
'''
vm_cls = QubesVmClasses[vm_type]
if 'template' in kwargs:
if not vm_cls.is_template_compatible(kwargs['template']):
raise QubesException(
'Template not compatible with selected VM type')
vm = vm_cls(qid=qid, collection=self, **kwargs)
if not self.verify_new_vm(vm):
raise QubesException("Wrong VM description!")
self[vm.qid] = vm
# make first created NetVM the default one
if self.default_fw_netvm_qid is None and vm.is_netvm():
self.set_default_fw_netvm(vm)
if self.default_netvm_qid is None and vm.is_proxyvm():
self.set_default_netvm(vm)
# make first created TemplateVM the default one
if self.default_template_qid is None and vm.is_template():
self.set_default_template(vm)
# make first created ProxyVM the UpdateVM
if self.updatevm_qid is None and vm.is_proxyvm():
self.set_updatevm_vm(vm)
# by default ClockVM is the first NetVM
if self.clockvm_qid is None and vm.is_netvm():
self.set_clockvm_vm(vm)
return vm
def __xml__(self):
element = lxml.etree.Element('domain')
element.set('id', 'domain-' + str(self.qid))
@ -439,7 +401,7 @@ class BaseVM(qubes.PropertyHolder):
conf_appvm = open(file_path, "w")
conf_appvm.write(domain_config)
conf_appvm.close()
except:
except: # pylint: disable=bare-except
# Ignore errors
pass
finally:
@ -503,7 +465,7 @@ class BaseVM(qubes.PropertyHolder):
tree.write(fd, encoding="UTF-8", pretty_print=True)
fd.close()
os.umask(old_umask)
except EnvironmentError as err:
except EnvironmentError as err: # pylint: disable=broad-except
print >> sys.stderr, "{0}: save error: {1}".format(
os.path.basename(sys.argv[0]), err)
return False
@ -590,7 +552,7 @@ class BaseVM(qubes.PropertyHolder):
conf["rules"].append(rule)
except EnvironmentError as err:
except EnvironmentError as err: # pylint: disable=broad-except
# problem accessing file, like ENOTFOUND, EPERM or sth
# return default config
return conf

View File

@ -40,6 +40,7 @@ class AdminVM(qubes.vm.qubesvm.QubesVM):
default=None,
doc='There are other ways to set kernel for Dom0.')
@property
def xid(self):
'''Always ``0``.
@ -49,6 +50,7 @@ class AdminVM(qubes.vm.qubesvm.QubesVM):
'''
return 0
@property
def libvirt_domain(self):
'''Always :py:obj:`None`.
@ -59,16 +61,6 @@ class AdminVM(qubes.vm.qubesvm.QubesVM):
return None
@property
def kernels_dir(self):
'''Always :py:obj:`None`.
.. seealso:
:py:attr:`qubes.vm.qubesvm.QubesVM.kernels_dir`
'''
return None
# XXX probably unneeded, will return None as we don't have netvm
# @property
# def ip(self):

View File

@ -17,6 +17,7 @@ class AppVM(qubes.vm.qubesvm.QubesVM):
@qubes.events.handler('domain-loaded')
def on_domain_loaded(self, event):
# pylint: disable=unused-argument
# Some additional checks for template based VM
assert self.template
#self.template.appvms.add(self) # XXX

View File

@ -51,8 +51,7 @@ import qubes.tools.qvm_ls
qmemman_present = False
try:
# pylint: disable=import-error
import qubes.qmemman_client
import qubes.qmemman.client
qmemman_present = True
except ImportError:
pass
@ -108,6 +107,7 @@ def _setter_kernel(self, prop, value):
def _setter_label(self, prop, value):
# pylint: disable=unused-argument
if isinstance(value, qubes.Label):
return value
if value.startswith('label-'):
@ -473,7 +473,7 @@ class QubesVM(qubes.vm.BaseVM):
def __init__(self, app, xml, **kwargs):
super(QubesVM, self).__init__(app, xml, **kwargs)
import qubes.vm.adminvm
import qubes.vm.adminvm # pylint: disable=redefined-outer-name
#Init private attrs
@ -521,6 +521,7 @@ class QubesVM(qubes.vm.BaseVM):
@qubes.events.handler('domain-init', 'domain-loaded')
def on_domain_init_loaded(self, event):
# pylint: disable=unused-argument
if not hasattr(self, 'uuid'):
self.uuid = uuid.uuid4()
@ -531,7 +532,7 @@ class QubesVM(qubes.vm.BaseVM):
if self.icon_path:
try:
os.remove(self.icon_path)
except:
except OSError:
pass
if hasattr(os, "symlink"):
os.symlink(new_label.icon_path, self.icon_path)
@ -724,7 +725,7 @@ class QubesVM(qubes.vm.BaseVM):
if mem_required is None:
mem_required = int(self.memory) * 1024 * 1024
if qmemman_present:
qmemman_client = qubes.qmemman_client.QMemmanClient()
qmemman_client = qubes.qmemman.client.QMemmanClient()
try:
got_memory = qmemman_client.request_memory(mem_required)
except IOError as e:
@ -1025,8 +1026,8 @@ class QubesVM(qubes.vm.BaseVM):
subprocess.call(
[qubes.config.system_path['monitor_layout_notify_cmd'],
self.name])
except Exception as e:
self.log.error('ERROR: {!s}'.format(e))
except Exception as e: # pylint: disable=broad-except
self.log.error('error starting gui-daemon: {!s}'.format(e))
self.wait_for_session()
@ -1473,7 +1474,7 @@ class QubesVM(qubes.vm.BaseVM):
.. seealso:: :py:meth:`get_root_img_sz`
'''
return qubes.utils.get_disk_usage(self.root_img)
return qubes.storage.get_disk_usage(self.root_img)
# XXX shouldn't this go only to vms that have root image?
@ -1507,7 +1508,7 @@ class QubesVM(qubes.vm.BaseVM):
.. seealso:: :py:meth:`get_private_img_sz`
''' # pylint: disable=invalid-name
return qubes.utils.get_disk_usage(self.private_img)
return qubes.storage.get_disk_usage(self.private_img)
def get_private_img_sz(self):
@ -1533,7 +1534,7 @@ class QubesVM(qubes.vm.BaseVM):
:rtype: FIXME
'''
return qubes.utils.get_disk_usage(self.dir_path)
return qubes.storage.get_disk_usage(self.dir_path)
# TODO move to storage
@ -1588,11 +1589,11 @@ class QubesVM(qubes.vm.BaseVM):
:returns: :py:obj:`True` if is outdated, :py:obj:`False` otherwise.
:rtype: bool
'''
# pylint: disable=no-member
# Makes sense only on VM based on template
if self.template is None:
return False
# pylint: disable=no-member
if not self.is_running():
return False
@ -1652,6 +1653,8 @@ class QubesVM(qubes.vm.BaseVM):
def create_qdb_entries(self):
'''Create entries in Qubes DB.
'''
# pylint: disable=no-member
self.qdb.write("/name", self.name)
self.qdb.write("/qubes-vm-type", self.__class__.__name__)
self.qdb.write("/qubes-vm-updateable", str(self.updateable))
@ -1753,12 +1756,16 @@ class QubesVM(qubes.vm.BaseVM):
# TODO: qmemman is still xen specific
untrusted_meminfo_key = self.app.vmm.xs.read('',
'/local/domain/{}/memory/meminfo'.format(self.xid))
if untrusted_meminfo_key is None or untrusted_meminfo_key == '':
return 0
domain = qmemman.DomainState(self.xid)
qmemman_algo.refresh_meminfo_for_domain(domain, untrusted_meminfo_key)
domain.memory_maximum = self.get_mem_static_max()*1024
return qmemman_algo.prefmem(domain)/1024
domain = qubes.qmemman.DomainState(self.xid)
qubes.qmemman.algo.refresh_meminfo_for_domain(
domain, untrusted_meminfo_key)
domain.memory_maximum = self.get_mem_static_max() * 1024
return qubes.qmemman.algo.prefmem(domain) / 1024

View File

@ -1,8 +1,6 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
import os.path
import qubes
import qubes.config
import qubes.vm.qubesvm
@ -27,7 +25,7 @@ class TemplateVM(qubes.vm.qubesvm.QubesVM):
def clone_disk_files(self, src):
super(QubesTemplateVm, self).clone_disk_files(src)
super(TemplateVM, self).clone_disk_files(src)
# Create root-cow.img
self.commit_changes()
@ -44,8 +42,3 @@ class TemplateVM(qubes.vm.qubesvm.QubesVM):
self.log.info(
'Commiting template update; COW: {}'.format(self.rootcow_img))
self.storage.commit_template_changes()
@property
def rootcow_img(self):
return os.path.join(self.dir_path, qubes.config.vm_files['rootcow_img'])