qubes: pylint fixes (disable unfounded messages)
Cette révision appartient à :
		
							Parent
							
								
									ac1739df6b
								
							
						
					
					
						révision
						bf29d5e5b5
					
				@ -61,6 +61,7 @@ import qubes.ext
 | 
			
		||||
if os.name == 'posix':
 | 
			
		||||
    import fcntl
 | 
			
		||||
elif os.name == 'nt':
 | 
			
		||||
    # pylint: disable=import-error
 | 
			
		||||
    import win32con
 | 
			
		||||
    import win32file
 | 
			
		||||
    import pywintypes
 | 
			
		||||
@ -187,6 +188,7 @@ class QubesHost(object):
 | 
			
		||||
        if self._no_cpus is not None:
 | 
			
		||||
            return
 | 
			
		||||
 | 
			
		||||
        # pylint: disable=unused-variable
 | 
			
		||||
        (model, memory, cpus, mhz, nodes, socket, cores, threads) = \
 | 
			
		||||
            self.app.vmm.libvirt_conn.getInfo()
 | 
			
		||||
        self._total_mem = long(memory) * 1024
 | 
			
		||||
@ -525,7 +527,7 @@ class VMCollection(object):
 | 
			
		||||
        raise LookupError("Cannot find unused netid!")
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class property(object):
 | 
			
		||||
class property(object): # pylint: disable=redefined-builtin,invalid-name
 | 
			
		||||
    '''Qubes property.
 | 
			
		||||
 | 
			
		||||
    This class holds one property that can be saved to and loaded from
 | 
			
		||||
@ -581,6 +583,7 @@ class property(object):
 | 
			
		||||
    def __init__(self, name, setter=None, saver=None, type=None,
 | 
			
		||||
            default=_NO_DEFAULT, load_stage=2, order=0, save_via_ref=False,
 | 
			
		||||
            doc=None):
 | 
			
		||||
        # pylint: disable=redefined-builtin
 | 
			
		||||
        self.__name__ = name
 | 
			
		||||
        self._setter = setter
 | 
			
		||||
        self._saver = saver if saver is not None else (
 | 
			
		||||
@ -638,7 +641,7 @@ class property(object):
 | 
			
		||||
        else:
 | 
			
		||||
            instance.fire_event_pre('property-pre-set:' + self.__name__, value)
 | 
			
		||||
 | 
			
		||||
        instance._init_property(self, value)
 | 
			
		||||
        instance._init_property(self, value) # pylint: disable=protected-access
 | 
			
		||||
 | 
			
		||||
        if has_oldvalue:
 | 
			
		||||
            instance.fire_event(
 | 
			
		||||
@ -689,6 +692,7 @@ class property(object):
 | 
			
		||||
        if not self.__doc__:
 | 
			
		||||
            return ''
 | 
			
		||||
 | 
			
		||||
        # pylint: disable=unused-variable
 | 
			
		||||
        output, pub = docutils.core.publish_programmatically(
 | 
			
		||||
            source_class=docutils.io.StringInput,
 | 
			
		||||
            source=' '.join(self.__doc__.strip().split()),
 | 
			
		||||
@ -716,6 +720,7 @@ class property(object):
 | 
			
		||||
    @staticmethod
 | 
			
		||||
    def dontsave(self, prop, value):
 | 
			
		||||
        '''Dummy saver that never saves anything.'''
 | 
			
		||||
        # pylint: disable=bad-staticmethod-argument,unused-argument
 | 
			
		||||
        raise DontSave()
 | 
			
		||||
 | 
			
		||||
    #
 | 
			
		||||
@ -730,7 +735,7 @@ class property(object):
 | 
			
		||||
        unwanted property. When someone attempts to load such a property, it
 | 
			
		||||
 | 
			
		||||
        :throws AttributeError: always
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=bad-staticmethod-argument,unused-argument
 | 
			
		||||
 | 
			
		||||
        raise AttributeError(
 | 
			
		||||
            'setting {} property on {} instance is forbidden'.format(
 | 
			
		||||
@ -744,7 +749,7 @@ class property(object):
 | 
			
		||||
        It accepts (case-insensitive) ``'0'``, ``'no'`` and ``false`` as
 | 
			
		||||
        :py:obj:`False` and ``'1'``, ``'yes'`` and ``'true'`` as
 | 
			
		||||
        :py:obj:`True`.
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=bad-staticmethod-argument,unused-argument
 | 
			
		||||
 | 
			
		||||
        lcvalue = value.lower()
 | 
			
		||||
        if lcvalue in ('0', 'no', 'false'):
 | 
			
		||||
@ -841,6 +846,7 @@ class PropertyHolder(qubes.events.Emitter):
 | 
			
		||||
        :param value: value
 | 
			
		||||
        '''
 | 
			
		||||
 | 
			
		||||
        # pylint: disable=protected-access
 | 
			
		||||
        setattr(self, self.get_property_def(prop)._attr_name, value)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -856,6 +862,7 @@ class PropertyHolder(qubes.events.Emitter):
 | 
			
		||||
        :rtype: bool
 | 
			
		||||
        '''
 | 
			
		||||
 | 
			
		||||
        # pylint: disable=protected-access
 | 
			
		||||
        return hasattr(self, self.get_property_def(prop)._attr_name)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -920,6 +927,7 @@ class PropertyHolder(qubes.events.Emitter):
 | 
			
		||||
        properties = lxml.etree.Element('properties')
 | 
			
		||||
 | 
			
		||||
        for prop in self.get_props_list():
 | 
			
		||||
            # pylint: disable=protected-access
 | 
			
		||||
            try:
 | 
			
		||||
                value = getattr(
 | 
			
		||||
                    self, (prop.__name__ if with_defaults else prop._attr_name))
 | 
			
		||||
@ -958,6 +966,7 @@ class PropertyHolder(qubes.events.Emitter):
 | 
			
		||||
 | 
			
		||||
        for prop in self.proplist():
 | 
			
		||||
            try:
 | 
			
		||||
                # pylint: disable=protected-access
 | 
			
		||||
                self._init_property(self, prop, getattr(src, prop._attr_name))
 | 
			
		||||
            except AttributeError:
 | 
			
		||||
                continue
 | 
			
		||||
@ -984,6 +993,7 @@ class PropertyHolder(qubes.events.Emitter):
 | 
			
		||||
            if value is None and not allow_none:
 | 
			
		||||
                raise AttributeError()
 | 
			
		||||
        except AttributeError:
 | 
			
		||||
            # pylint: disable=no-member
 | 
			
		||||
            msg = 'Required property {!r} not set on {!r}'.format(prop, self)
 | 
			
		||||
            if hard:
 | 
			
		||||
                raise AssertionError(msg)
 | 
			
		||||
@ -1119,8 +1129,9 @@ class Qubes(PropertyHolder):
 | 
			
		||||
 | 
			
		||||
        self.log = logging.getLogger('app')
 | 
			
		||||
 | 
			
		||||
        self._extensions = set(ext(self)
 | 
			
		||||
                               for ext in qubes.ext.Extension.register.values())
 | 
			
		||||
        # pylint: disable=no-member
 | 
			
		||||
        self._extensions = set(
 | 
			
		||||
            ext(self) for ext in qubes.ext.Extension.register.values())
 | 
			
		||||
 | 
			
		||||
        #: collection of all VMs managed by this Qubes instance
 | 
			
		||||
        self.domains = VMCollection(self)
 | 
			
		||||
@ -1166,6 +1177,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
        if os.name == 'posix':
 | 
			
		||||
            fcntl.lockf(self._storefd, fcntl.LOCK_EX)
 | 
			
		||||
        elif os.name == 'nt':
 | 
			
		||||
            # pylint: disable=protected-access
 | 
			
		||||
            win32file.LockFileEx(
 | 
			
		||||
                win32file._get_osfhandle(self._storefd.fileno()),
 | 
			
		||||
                win32con.LOCKFILE_EXCLUSIVE_LOCK,
 | 
			
		||||
@ -1304,6 +1316,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('domain-pre-deleted')
 | 
			
		||||
    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)
 | 
			
		||||
            if appvms:
 | 
			
		||||
@ -1315,6 +1328,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('domain-deleted')
 | 
			
		||||
    def on_domain_deleted(self, event, vm):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if self.default_netvm == vm:
 | 
			
		||||
            del self.default_netvm
 | 
			
		||||
        if self.default_fw_netvm == vm:
 | 
			
		||||
@ -1331,6 +1345,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-pre-set:clockvm')
 | 
			
		||||
    def on_property_pre_set_clockvm(self, event, name, newvalue, oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument,no-self-use
 | 
			
		||||
        if 'ntpd' in newvalue.services:
 | 
			
		||||
            if newvalue.services['ntpd']:
 | 
			
		||||
                raise QubesException('Cannot set {!r} as {!r} property since '
 | 
			
		||||
@ -1342,6 +1357,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
    @qubes.events.handler('property-pre-set:default_netvm')
 | 
			
		||||
    def on_property_pre_set_default_netvm(self, event, name, newvalue,
 | 
			
		||||
            oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument,invalid-name
 | 
			
		||||
        if newvalue is not None and oldvalue is not None \
 | 
			
		||||
                and oldvalue.is_running() and not newvalue.is_running() \
 | 
			
		||||
                and self.domains.get_vms_connected_to(oldvalue):
 | 
			
		||||
@ -1352,6 +1368,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
    @qubes.events.handler('property-set:default_fw_netvm')
 | 
			
		||||
    def on_property_set_default_netvm(self, event, name, newvalue,
 | 
			
		||||
            oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument,invalid-name
 | 
			
		||||
        for vm in self.domains:
 | 
			
		||||
            if not vm.provides_network and vm.property_is_default('netvm'):
 | 
			
		||||
                # fire property-del:netvm as it is responsible for resetting
 | 
			
		||||
@ -1362,6 +1379,7 @@ class Qubes(PropertyHolder):
 | 
			
		||||
    @qubes.events.handler('property-set:default_netvm')
 | 
			
		||||
    def on_property_set_default_netvm(self, event, name, newvalue,
 | 
			
		||||
            oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        for vm in self.domains:
 | 
			
		||||
            if vm.provides_network and vm.property_is_default('netvm'):
 | 
			
		||||
                # fire property-del:netvm as it is responsible for resetting
 | 
			
		||||
 | 
			
		||||
@ -1,5 +1,6 @@
 | 
			
		||||
#!/usr/bin/python2 -O
 | 
			
		||||
# vim: fileencoding=utf-8
 | 
			
		||||
# pylint: disable=wildcard-import,unused-wildcard-import
 | 
			
		||||
 | 
			
		||||
from qubes.vm import *
 | 
			
		||||
from qubes.ext import *
 | 
			
		||||
 | 
			
		||||
@ -69,7 +69,7 @@ def ticket(name, rawtext, text, lineno, inliner, options={}, content=[]):
 | 
			
		||||
        that called this function
 | 
			
		||||
    :param options: Directive options for customisation
 | 
			
		||||
    :param content: The directive content for customisation
 | 
			
		||||
    '''
 | 
			
		||||
    ''' # pylint: disable=unused-argument
 | 
			
		||||
 | 
			
		||||
    ticket = text.lstrip('#')
 | 
			
		||||
    if not ticket.isdigit():
 | 
			
		||||
@ -100,6 +100,7 @@ def ticket(name, rawtext, text, lineno, inliner, options={}, content=[]):
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class versioncheck(docutils.nodes.warning):
 | 
			
		||||
    # pylint: disable=invalid-name
 | 
			
		||||
    pass
 | 
			
		||||
 | 
			
		||||
def visit(self, node):
 | 
			
		||||
@ -150,6 +151,7 @@ class VersionCheck(docutils.parsers.rst.Directive):
 | 
			
		||||
event_sig_re = re.compile(r'([a-zA-Z-:<>]+)\s*\((.*)\)')
 | 
			
		||||
 | 
			
		||||
def parse_event(env, sig, signode):
 | 
			
		||||
    # pylint: disable=unused-argument
 | 
			
		||||
    m = event_sig_re.match(sig)
 | 
			
		||||
    if not m:
 | 
			
		||||
        signode += sphinx.addnodes.desc_name(sig, sig)
 | 
			
		||||
 | 
			
		||||
@ -110,6 +110,7 @@ class Emitter(object):
 | 
			
		||||
        :param collections.Callable handler: handler callable
 | 
			
		||||
        '''
 | 
			
		||||
 | 
			
		||||
        # pylint: disable=no-member
 | 
			
		||||
        cls.__handlers__[event].add(handler)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -50,7 +50,7 @@ class Extension(object):
 | 
			
		||||
    '''Base class for all extensions
 | 
			
		||||
 | 
			
		||||
    :param qubes.Qubes app: application object
 | 
			
		||||
    '''
 | 
			
		||||
    ''' # pylint: disable=too-few-public-methods
 | 
			
		||||
 | 
			
		||||
    __metaclass__ = ExtensionPlugin
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -33,6 +33,7 @@ import os
 | 
			
		||||
class Plugin(type):
 | 
			
		||||
    '''Base metaclass for plugins'''
 | 
			
		||||
    def __init__(cls, name, bases, dict_):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if hasattr(cls, 'register'):
 | 
			
		||||
            cls.register[cls.__name__] = cls
 | 
			
		||||
        else:
 | 
			
		||||
@ -49,13 +50,18 @@ def load(modfile):
 | 
			
		||||
 | 
			
		||||
    >>> __all__ = qubes.plugins.load(__file__) # doctest: +SKIP
 | 
			
		||||
    '''
 | 
			
		||||
 | 
			
		||||
    path = os.path.dirname(modfile)
 | 
			
		||||
    listdir = os.listdir(path)
 | 
			
		||||
    ret = set()
 | 
			
		||||
 | 
			
		||||
    # pylint: disable=unused-variable
 | 
			
		||||
    for suffix, mode, type_ in imp.get_suffixes():
 | 
			
		||||
        for filename in listdir:
 | 
			
		||||
            if filename.endswith(suffix):
 | 
			
		||||
                ret.add(filename[:-len(suffix)])
 | 
			
		||||
 | 
			
		||||
    if '__init__' in ret:
 | 
			
		||||
        ret.remove('__init__')
 | 
			
		||||
 | 
			
		||||
    return list(sorted(ret))
 | 
			
		||||
 | 
			
		||||
@ -161,6 +161,7 @@ class Element(object):
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class Schema(object):
 | 
			
		||||
    # pylint: disable=too-few-public-methods
 | 
			
		||||
    nsmap = {
 | 
			
		||||
        'rng': 'http://relaxng.org/ns/structure/1.0',
 | 
			
		||||
        'q': 'http://qubes-os.org/qubes/3',
 | 
			
		||||
 | 
			
		||||
@ -42,7 +42,7 @@ class VMStorage(object):
 | 
			
		||||
 | 
			
		||||
    This is base class for all other implementations, mostly with Xen on Linux
 | 
			
		||||
    in mind.
 | 
			
		||||
    '''
 | 
			
		||||
    ''' # pylint: disable=abstract-class-little-used
 | 
			
		||||
 | 
			
		||||
    def __init__(self, vm, private_img_size=None, root_img_size=None,
 | 
			
		||||
            modules_img=None, modules_img_rw=False):
 | 
			
		||||
@ -116,6 +116,7 @@ class VMStorage(object):
 | 
			
		||||
        return qubes.utils.get_disk_usage(self.vmdir)
 | 
			
		||||
 | 
			
		||||
    def get_disk_utilization_private_img(self):
 | 
			
		||||
        # pylint: disable=invalid-name
 | 
			
		||||
        return qubes.utils.get_disk_usage(self.private_img)
 | 
			
		||||
 | 
			
		||||
    def get_private_img_sz(self):
 | 
			
		||||
 | 
			
		||||
@ -49,6 +49,7 @@ class XenVMStorage(qubes.storage.VMStorage):
 | 
			
		||||
    modules_dev = 'xvdd'
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
    # pylint: disable=redefined-builtin
 | 
			
		||||
    @staticmethod
 | 
			
		||||
    def _format_disk_dev(path, vdev, script=None, rw=True, type='disk',
 | 
			
		||||
            domain=None):
 | 
			
		||||
 | 
			
		||||
@ -64,7 +64,7 @@ def skipUnlessDom0(test_item):
 | 
			
		||||
 | 
			
		||||
    Some tests (especially integration tests) have to be run in more or less
 | 
			
		||||
    working dom0. This is checked by connecting to libvirt.
 | 
			
		||||
    '''
 | 
			
		||||
    ''' # pylint: disable=invalid-name
 | 
			
		||||
 | 
			
		||||
    return unittest.skipUnless(in_dom0, 'outside dom0')(test_item)
 | 
			
		||||
 | 
			
		||||
@ -74,7 +74,7 @@ def skipUnlessGit(test_item):
 | 
			
		||||
 | 
			
		||||
    There are very few tests that an be run only in git. One example is
 | 
			
		||||
    correctness of example code that won't get included in RPM.
 | 
			
		||||
    '''
 | 
			
		||||
    ''' # pylint: disable=invalid-name
 | 
			
		||||
 | 
			
		||||
    return unittest.skipUnless(in_git, 'outside git tree')(test_item)
 | 
			
		||||
 | 
			
		||||
@ -129,7 +129,8 @@ class QubesTestCase(unittest.TestCase):
 | 
			
		||||
        :param xml2: second element
 | 
			
		||||
        :type xml1: :py:class:`lxml.etree._Element`
 | 
			
		||||
        :type xml2: :py:class:`lxml.etree._Element`
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=invalid-name
 | 
			
		||||
 | 
			
		||||
        self.assertEqual(xml1.tag, xml2.tag)
 | 
			
		||||
        self.assertEqual(xml1.text, xml2.text)
 | 
			
		||||
        self.assertItemsEqual(xml1.keys(), xml2.keys())
 | 
			
		||||
@ -148,7 +149,7 @@ class QubesTestCase(unittest.TestCase):
 | 
			
		||||
            an event
 | 
			
		||||
        :param list kwargs: when given, all items must appear in kwargs passed \
 | 
			
		||||
            to an event
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=invalid-name
 | 
			
		||||
 | 
			
		||||
        for ev, ev_args, ev_kwargs in emitter.fired_events:
 | 
			
		||||
            if ev != event:
 | 
			
		||||
@ -173,7 +174,7 @@ class QubesTestCase(unittest.TestCase):
 | 
			
		||||
            an event
 | 
			
		||||
        :param list kwargs: when given, all items must appear in kwargs passed \
 | 
			
		||||
            to an event
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=invalid-name
 | 
			
		||||
 | 
			
		||||
        for ev, ev_args, ev_kwargs in emitter.fired_events:
 | 
			
		||||
            if ev != event:
 | 
			
		||||
@ -202,12 +203,13 @@ class QubesTestCase(unittest.TestCase):
 | 
			
		||||
        :param lxml.etree._Element xml: XML element instance to check
 | 
			
		||||
        :param str file: filename of Relax NG schema
 | 
			
		||||
        :param str schema: optional explicit schema string
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=invalid-name,redefined-builtin
 | 
			
		||||
 | 
			
		||||
        if schema is not None and file is None:
 | 
			
		||||
            relaxng = schema
 | 
			
		||||
            if isinstance(relaxng, str):
 | 
			
		||||
                relaxng = lxml.etree.XML(relaxng)
 | 
			
		||||
            # pylint: disable=protected-access
 | 
			
		||||
            if isinstance(relaxng, lxml.etree._Element):
 | 
			
		||||
                relaxng = lxml.etree.RelaxNG(relaxng)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -31,6 +31,7 @@ class TC_00_Emitter(qubes.tests.QubesTestCase):
 | 
			
		||||
        testevent_fired = [False]
 | 
			
		||||
 | 
			
		||||
        def on_testevent(subject, event):
 | 
			
		||||
            # pylint: disable=unused-argument
 | 
			
		||||
            if event == 'testevent':
 | 
			
		||||
                testevent_fired[0] = True
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -1,5 +1,6 @@
 | 
			
		||||
#!/usr/bin/python2 -O
 | 
			
		||||
# vim: fileencoding=utf-8
 | 
			
		||||
# pylint: disable=protected-access,pointless-statement
 | 
			
		||||
 | 
			
		||||
#
 | 
			
		||||
# The Qubes OS Project, https://www.qubes-os.org/
 | 
			
		||||
@ -67,7 +68,7 @@ class TC_10_property(qubes.tests.QubesTestCase):
 | 
			
		||||
        try:
 | 
			
		||||
            class TestHolder(qubes.tests.TestEmitter, qubes.PropertyHolder):
 | 
			
		||||
                testprop1 = qubes.property('testprop1')
 | 
			
		||||
        except:
 | 
			
		||||
        except: # pylint: disable=bare-except
 | 
			
		||||
            self.skipTest('TestHolder class definition failed')
 | 
			
		||||
        self.holder = TestHolder(None)
 | 
			
		||||
 | 
			
		||||
@ -348,6 +349,7 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
 | 
			
		||||
class TC_90_Qubes(qubes.tests.QubesTestCase):
 | 
			
		||||
    @qubes.tests.skipUnlessDom0
 | 
			
		||||
    def test_000_init_empty(self):
 | 
			
		||||
        # pylint: disable=no-self-use,unused-variable,bare-except
 | 
			
		||||
        try:
 | 
			
		||||
            os.unlink('/tmp/qubestest.xml')
 | 
			
		||||
        except:
 | 
			
		||||
 | 
			
		||||
@ -45,6 +45,7 @@ class ANSIColor(dict):
 | 
			
		||||
        except curses.error:
 | 
			
		||||
            return
 | 
			
		||||
 | 
			
		||||
        # pylint: disable=bad-whitespace
 | 
			
		||||
        self['black']   = curses.tparm(curses.tigetstr('setaf'), 0)
 | 
			
		||||
        self['red']     = curses.tparm(curses.tigetstr('setaf'), 1)
 | 
			
		||||
        self['green']   = curses.tparm(curses.tigetstr('setaf'), 2)
 | 
			
		||||
@ -58,6 +59,7 @@ class ANSIColor(dict):
 | 
			
		||||
        self['normal']  = curses.tigetstr('sgr0')
 | 
			
		||||
 | 
			
		||||
    def __missing__(self, key):
 | 
			
		||||
        # pylint: disable=unused-argument,no-self-use
 | 
			
		||||
        return ''
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -73,7 +75,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
    def __init__(self, stream, descriptions, verbosity):
 | 
			
		||||
        super(ANSITestResult, self).__init__(stream, descriptions, verbosity)
 | 
			
		||||
        self.stream = stream
 | 
			
		||||
        self.showAll = verbosity > 1
 | 
			
		||||
        self.showAll = verbosity > 1 # pylint: disable=invalid-name
 | 
			
		||||
        self.dots = verbosity == 1
 | 
			
		||||
        self.descriptions = descriptions
 | 
			
		||||
 | 
			
		||||
@ -88,7 +90,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
            return '{color[bold]}{}{color[normal]}'.format(
 | 
			
		||||
                err[0].__name__, color=self.color)
 | 
			
		||||
 | 
			
		||||
    def getDescription(self, test):
 | 
			
		||||
    def getDescription(self, test): # pylint: disable=invalid-name
 | 
			
		||||
        teststr = str(test).split('/')
 | 
			
		||||
        teststr[-1] = '{color[bold]}{}{color[normal]}'.format(
 | 
			
		||||
            teststr[-1], color=self.color)
 | 
			
		||||
@ -101,14 +103,14 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
        else:
 | 
			
		||||
            return teststr
 | 
			
		||||
 | 
			
		||||
    def startTest(self, test):
 | 
			
		||||
    def startTest(self, test): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).startTest(test)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.write(self.getDescription(test))
 | 
			
		||||
            self.stream.write(' ... ')
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def addSuccess(self, test):
 | 
			
		||||
    def addSuccess(self, test): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).addSuccess(test)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.writeln('{color[green]}ok{color[normal]}'.format(
 | 
			
		||||
@ -117,7 +119,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
            self.stream.write('.')
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def addError(self, test, err):
 | 
			
		||||
    def addError(self, test, err): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).addError(test, err)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.writeln(
 | 
			
		||||
@ -129,7 +131,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
                    color=self.color))
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def addFailure(self, test, err):
 | 
			
		||||
    def addFailure(self, test, err): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).addFailure(test, err)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.writeln('{color[red]}FAIL{color[normal]}'.format(
 | 
			
		||||
@ -139,7 +141,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
                color=self.color))
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def addSkip(self, test, reason):
 | 
			
		||||
    def addSkip(self, test, reason): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).addSkip(test, reason)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.writeln(
 | 
			
		||||
@ -150,7 +152,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
                color=self.color))
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def addExpectedFailure(self, test, err):
 | 
			
		||||
    def addExpectedFailure(self, test, err): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).addExpectedFailure(test, err)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.writeln(
 | 
			
		||||
@ -161,7 +163,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
                color=self.color))
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def addUnexpectedSuccess(self, test):
 | 
			
		||||
    def addUnexpectedSuccess(self, test): # pylint: disable=invalid-name
 | 
			
		||||
        super(ANSITestResult, self).addUnexpectedSuccess(test)
 | 
			
		||||
        if self.showAll:
 | 
			
		||||
            self.stream.writeln(
 | 
			
		||||
@ -173,7 +175,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
                    color=self.color))
 | 
			
		||||
            self.stream.flush()
 | 
			
		||||
 | 
			
		||||
    def printErrors(self):
 | 
			
		||||
    def printErrors(self): # pylint: disable=invalid-name
 | 
			
		||||
        if self.dots or self.showAll:
 | 
			
		||||
            self.stream.writeln()
 | 
			
		||||
        self.printErrorList(
 | 
			
		||||
@ -184,7 +186,7 @@ class ANSITestResult(unittest.TestResult):
 | 
			
		||||
            '{color[red]}FAIL{color[normal]}'.format(color=self.color),
 | 
			
		||||
            self.failures)
 | 
			
		||||
 | 
			
		||||
    def printErrorList(self, flavour, errors):
 | 
			
		||||
    def printErrorList(self, flavour, errors): # pylint: disable=invalid-name
 | 
			
		||||
        for test, err in errors:
 | 
			
		||||
            self.stream.writeln(self.separator1)
 | 
			
		||||
            self.stream.writeln('%s: %s' % (flavour, self.getDescription(test)))
 | 
			
		||||
@ -196,6 +198,7 @@ def demo(verbosity=2):
 | 
			
		||||
    import qubes.tests
 | 
			
		||||
    class TC_Demo(qubes.tests.QubesTestCase):
 | 
			
		||||
        '''Demo class'''
 | 
			
		||||
        # pylint: disable=no-self-use
 | 
			
		||||
        def test_0_success(self):
 | 
			
		||||
            '''Demo test (success)'''
 | 
			
		||||
            pass
 | 
			
		||||
 | 
			
		||||
@ -30,10 +30,12 @@ import qubes.vm.adminvm
 | 
			
		||||
import qubes.tests
 | 
			
		||||
 | 
			
		||||
class TestVMM(object):
 | 
			
		||||
    # pylint: disable=too-few-public-methods
 | 
			
		||||
    def __init__(self, offline_mode=False):
 | 
			
		||||
        self.offline_mode = offline_mode
 | 
			
		||||
 | 
			
		||||
class TestHost(object):
 | 
			
		||||
    # pylint: disable=too-few-public-methods
 | 
			
		||||
    def __init__(self, offline_mode=False):
 | 
			
		||||
        self.memory_total = 1000
 | 
			
		||||
 | 
			
		||||
@ -51,7 +53,7 @@ class TC_00_AdminVM(qubes.tests.QubesTestCase):
 | 
			
		||||
            self.app = TestApp()
 | 
			
		||||
            self.vm = qubes.vm.adminvm.AdminVM(self.app,
 | 
			
		||||
                xml=None, qid=0, name='dom0')
 | 
			
		||||
        except:
 | 
			
		||||
        except: # pylint: disable=bare-except
 | 
			
		||||
            if self.id().endswith('.test_000_init'):
 | 
			
		||||
                raise
 | 
			
		||||
            self.skipTest('setup failed')
 | 
			
		||||
@ -85,6 +87,7 @@ class TC_00_AdminVM(qubes.tests.QubesTestCase):
 | 
			
		||||
        self.assertEqual(self.vm.get_disk_utilization(), 0)
 | 
			
		||||
 | 
			
		||||
    def test_305_get_disk_utilization_private_img(self):
 | 
			
		||||
        # pylint: disable=invalid-name
 | 
			
		||||
        self.assertEqual(self.vm.get_disk_utilization_private_img(), 0)
 | 
			
		||||
 | 
			
		||||
    def test_306_get_private_img_sz(self):
 | 
			
		||||
 | 
			
		||||
@ -1,5 +1,6 @@
 | 
			
		||||
#!/usr/bin/python2 -O
 | 
			
		||||
# vim: fileencoding=utf-8
 | 
			
		||||
# pylint: disable=protected-access
 | 
			
		||||
 | 
			
		||||
#
 | 
			
		||||
# The Qubes OS Project, https://www.qubes-os.org/
 | 
			
		||||
 | 
			
		||||
@ -1,5 +1,6 @@
 | 
			
		||||
#!/usr/bin/python2 -O
 | 
			
		||||
# vim: fileencoding=utf-8
 | 
			
		||||
# pylint: disable=protected-access
 | 
			
		||||
 | 
			
		||||
#
 | 
			
		||||
# The Qubes OS Project, https://www.qubes-os.org/
 | 
			
		||||
@ -31,10 +32,12 @@ import qubes.tests
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class TestProp(object):
 | 
			
		||||
    # pylint: disable=too-few-public-methods
 | 
			
		||||
    __name__ = 'testprop'
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class TestVM(object):
 | 
			
		||||
    # pylint: disable=too-few-public-methods
 | 
			
		||||
    def __init__(self):
 | 
			
		||||
        self.running = False
 | 
			
		||||
        self.installed_by_rpm = False
 | 
			
		||||
@ -68,14 +71,17 @@ class TC_00_setters(qubes.tests.QubesTestCase):
 | 
			
		||||
            'test_name-1')
 | 
			
		||||
 | 
			
		||||
    def test_011_setter_name_longer_than_31(self):
 | 
			
		||||
        # pylint: disable=invalid-name
 | 
			
		||||
        with self.assertRaises(ValueError):
 | 
			
		||||
            qubes.vm.qubesvm._setter_name(self.vm, self.prop, 't' * 32)
 | 
			
		||||
 | 
			
		||||
    def test_012_setter_name_illegal_character(self):
 | 
			
		||||
        # pylint: disable=invalid-name
 | 
			
		||||
        with self.assertRaises(ValueError):
 | 
			
		||||
            qubes.vm.qubesvm._setter_name(self.vm, self.prop, 'test#')
 | 
			
		||||
 | 
			
		||||
    def test_013_setter_name_first_not_letter(self):
 | 
			
		||||
        # pylint: disable=invalid-name
 | 
			
		||||
        with self.assertRaises(ValueError):
 | 
			
		||||
            qubes.vm.qubesvm._setter_name(self.vm, self.prop, '1test')
 | 
			
		||||
 | 
			
		||||
@ -85,6 +91,7 @@ class TC_00_setters(qubes.tests.QubesTestCase):
 | 
			
		||||
            qubes.vm.qubesvm._setter_name(self.vm, self.prop, 'testname')
 | 
			
		||||
 | 
			
		||||
    def test_015_setter_name_installed_by_rpm(self):
 | 
			
		||||
        # pylint: disable=invalid-name
 | 
			
		||||
        self.vm.installed_by_rpm = True
 | 
			
		||||
        with self.assertRaises(qubes.QubesException):
 | 
			
		||||
            qubes.vm.qubesvm._setter_name(self.vm, self.prop, 'testname')
 | 
			
		||||
 | 
			
		||||
@ -137,11 +137,13 @@ class BaseVM(qubes.PropertyHolder):
 | 
			
		||||
    provides basic framework. It contains no management logic. For that, see
 | 
			
		||||
    :py:class:`qubes.vm.qubesvm.QubesVM`.
 | 
			
		||||
    '''
 | 
			
		||||
    # pylint: disable=no-member
 | 
			
		||||
 | 
			
		||||
    __metaclass__ = BaseVMMeta
 | 
			
		||||
 | 
			
		||||
    def __init__(self, app, xml, load_stage=2, services={}, devices=None,
 | 
			
		||||
            tags={}, *args, **kwargs):
 | 
			
		||||
        # pylint: disable=redefined-outer-name
 | 
			
		||||
        #: mother :py:class:`qubes.Qubes` object
 | 
			
		||||
        self.app = app
 | 
			
		||||
 | 
			
		||||
@ -217,7 +219,7 @@ class BaseVM(qubes.PropertyHolder):
 | 
			
		||||
        :param lxml.etree._Element xml: XML node reference
 | 
			
		||||
        :param int load_stage: do not change the default (2) unless you know, \
 | 
			
		||||
            what you are doing
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=redefined-outer-name
 | 
			
		||||
 | 
			
		||||
        if xml is None:
 | 
			
		||||
            return cls(app)
 | 
			
		||||
 | 
			
		||||
@ -141,7 +141,7 @@ class AdminVM(qubes.vm.qubesvm.QubesVM):
 | 
			
		||||
 | 
			
		||||
        .. seealso:
 | 
			
		||||
           :py:meth:`qubes.vm.qubesvm.QubesVM.start`
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=unused-argument
 | 
			
		||||
        raise qubes.QubesException('Cannot start Dom0 fake domain!')
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -47,6 +47,7 @@ import qubes.vm
 | 
			
		||||
 | 
			
		||||
qmemman_present = False
 | 
			
		||||
try:
 | 
			
		||||
    # pylint: disable=import-error
 | 
			
		||||
    import qubes.qmemman_client
 | 
			
		||||
    qmemman_present = True
 | 
			
		||||
except ImportError:
 | 
			
		||||
@ -54,6 +55,7 @@ except ImportError:
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def _setter_qid(self, prop, value):
 | 
			
		||||
    # pylint: disable=unused-argument
 | 
			
		||||
    if not 0 <= value <= qubes.config.max_qid:
 | 
			
		||||
        raise ValueError(
 | 
			
		||||
            '{} value must be between 0 and qubes.config.max_qid'.format(
 | 
			
		||||
@ -85,6 +87,7 @@ def _setter_name(self, prop, value):
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def _setter_kernel(self, prop, value):
 | 
			
		||||
    # pylint: disable=unused-argument
 | 
			
		||||
    if not os.path.exists(os.path.join(system_path[
 | 
			
		||||
        'qubes_kernels_base_dir'], value)):
 | 
			
		||||
        raise qubes.QubesException('Kernel {!r} not installed'.format(value))
 | 
			
		||||
@ -381,6 +384,8 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
    @property
 | 
			
		||||
    def gateway(self):
 | 
			
		||||
        '''Gateway for other domains that use this domain as netvm.'''
 | 
			
		||||
        # pylint: disable=no-self-use
 | 
			
		||||
 | 
			
		||||
        # This is gateway IP for _other_ VMs, so make sense only in NetVMs
 | 
			
		||||
        return None
 | 
			
		||||
 | 
			
		||||
@ -458,6 +463,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-set:label')
 | 
			
		||||
    def on_property_set_label(self, event, name, new_label, old_label=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if self.icon_path:
 | 
			
		||||
            try:
 | 
			
		||||
                os.remove(self.icon_path)
 | 
			
		||||
@ -473,6 +479,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-del:netvm')
 | 
			
		||||
    def on_property_del_netvm(self, event, name, old_netvm):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        # we are changing to default netvm
 | 
			
		||||
        new_netvm = self.netvm
 | 
			
		||||
        if new_netvm == old_netvm:
 | 
			
		||||
@ -482,6 +489,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-set:netvm')
 | 
			
		||||
    def on_property_set_netvm(self, event, name, new_netvm, old_netvm=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if self.is_running() and new_netvm is not None \
 | 
			
		||||
                and not new_netvm.is_running():
 | 
			
		||||
            raise QubesException("Cannot dynamically attach to stopped NetVM")
 | 
			
		||||
@ -524,6 +532,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-pre-set:name')
 | 
			
		||||
    def on_property_pre_set_name(self, event, name, newvalue, oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        # TODO not self.is_stopped() would be more appropriate
 | 
			
		||||
        if self.is_running():
 | 
			
		||||
            raise QubesException('Cannot change name of running domain')
 | 
			
		||||
@ -531,6 +540,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-pre-set:dir_path')
 | 
			
		||||
    def on_property_pre_set_name(self, event, name, newvalue, oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        # TODO not self.is_stopped() would be more appropriate
 | 
			
		||||
        if self.is_running():
 | 
			
		||||
            raise QubesException('Cannot change dir_path of running domain')
 | 
			
		||||
@ -538,11 +548,13 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-set:dir_path')
 | 
			
		||||
    def on_property_set_dir_path(self, event, name, newvalue, oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        self.storage.rename(newvalue, oldvalue)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('property-set:name')
 | 
			
		||||
    def on_property_set_name(self, event, name, new_name, old_name=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if self._libvirt_domain is not None:
 | 
			
		||||
            self.libvirt_domain.undefine()
 | 
			
		||||
            self._libvirt_domain = None
 | 
			
		||||
@ -574,6 +586,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
    @qubes.events.handler('property-pre-set:autostart')
 | 
			
		||||
    def on_property_pre_set_autostart(self, event, prop, name, value,
 | 
			
		||||
            oldvalue=None):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if subprocess.call(['sudo', 'systemctl',
 | 
			
		||||
                ('enable' if value else 'disable'),
 | 
			
		||||
                'qubes-vm@{}.service'.format(self.name)]):
 | 
			
		||||
@ -582,6 +595,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('device-pre-attached:pci')
 | 
			
		||||
    def on_device_pre_attached_pci(self, event, pci):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if not os.path.exists('/sys/bus/pci/devices/0000:%s' % pci):
 | 
			
		||||
            raise QubesException("Invalid PCI device: %s" % pci)
 | 
			
		||||
        if not self.is_running():
 | 
			
		||||
@ -600,6 +614,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
    @qubes.events.handler('device-pre-detached:pci')
 | 
			
		||||
    def on_device_pre_detached_pci(self, event, pci):
 | 
			
		||||
        # pylint: disable=unused-argument
 | 
			
		||||
        if not self.is_running():
 | 
			
		||||
            return
 | 
			
		||||
 | 
			
		||||
@ -836,7 +851,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
                raise QubesException(
 | 
			
		||||
                    'Error while starting the {!r} VM: {!s}'.format(
 | 
			
		||||
                        self.name, e))
 | 
			
		||||
            except (MemoryError) as err:
 | 
			
		||||
            except MemoryError:
 | 
			
		||||
                raise QubesException('Not enough memory to start {!r} VM! '
 | 
			
		||||
                    'Close one or more running VMs and try again.'.format(
 | 
			
		||||
                        self.name))
 | 
			
		||||
@ -907,7 +922,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
        :param str user: username to run service as
 | 
			
		||||
        :param bool passio_popen: passed verbatim to :py:meth:`run`
 | 
			
		||||
        :param str input: string passed as input to service
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=redefined-builtin
 | 
			
		||||
 | 
			
		||||
        if input is not None and passio_popen is not None:
 | 
			
		||||
            raise ValueError("'input' and 'passio_popen' cannot be used "
 | 
			
		||||
@ -1021,6 +1036,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
        '''
 | 
			
		||||
 | 
			
		||||
        if source_template is None:
 | 
			
		||||
            # pylint: disable=no-member
 | 
			
		||||
            source_template = self.template
 | 
			
		||||
        assert source_template is not None
 | 
			
		||||
 | 
			
		||||
@ -1247,8 +1263,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
 | 
			
		||||
            https://libvirt.org/html/libvirt-libvirt-domain.html#virDomainState
 | 
			
		||||
                Libvirt API for changing state of a domain.
 | 
			
		||||
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=too-many-return-statements
 | 
			
		||||
 | 
			
		||||
        libvirt_domain = self.libvirt_domain
 | 
			
		||||
        if libvirt_domain is None:
 | 
			
		||||
@ -1435,7 +1450,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
        :rtype: FIXME
 | 
			
		||||
 | 
			
		||||
        .. seealso:: :py:meth:`get_private_img_sz`
 | 
			
		||||
        '''
 | 
			
		||||
        ''' # pylint: disable=invalid-name
 | 
			
		||||
 | 
			
		||||
        return qubes.utils.get_disk_usage(self.private_img)
 | 
			
		||||
 | 
			
		||||
@ -1520,6 +1535,7 @@ class QubesVM(qubes.vm.BaseVM):
 | 
			
		||||
        # 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
 | 
			
		||||
 | 
			
		||||
		Chargement…
	
		Référencer dans un nouveau ticket
	
	Block a user