qubes: port core to python3

fixes QubesOS/qubes-issues#2074
This commit is contained in:
Wojtek Porczyk 2017-01-18 22:16:46 +01:00
parent 0f1672dc63
commit d74567d65f
100 changed files with 222 additions and 417 deletions

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -30,9 +27,7 @@ Qubes OS
:copyright: © 2010-2015 Invisible Things Lab
'''
from __future__ import absolute_import
import __builtin__
import builtins
import collections
import os
import os.path
@ -110,7 +105,7 @@ class Label(object):
self.name)
@__builtin__.property
@builtins.property
def icon_path(self):
'''Icon path
@ -121,7 +116,7 @@ class Label(object):
self.icon) + ".png"
@__builtin__.property
@builtins.property
def icon_path_dispvm(self):
'''Icon path
@ -360,7 +355,7 @@ class property(object): # pylint: disable=redefined-builtin,invalid-name
:py:obj:`True`.
''' # pylint: disable=bad-staticmethod-argument,unused-argument
if isinstance(value, basestring):
if isinstance(value, str):
lcvalue = value.lower()
if lcvalue in ('0', 'no', 'false', 'off'):
return False

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -66,7 +63,6 @@ import qubes.vm.adminvm # pylint: disable=wrong-import-position
import qubes.vm.qubesvm # pylint: disable=wrong-import-position
import qubes.vm.templatevm # pylint: disable=wrong-import-position
class VirDomainWrapper(object):
# pylint: disable=too-few-public-methods
@ -241,7 +237,7 @@ class QubesHost(object):
# pylint: disable=unused-variable
(model, memory, cpus, mhz, nodes, socket, cores, threads) = \
self.app.vmm.libvirt_conn.getInfo()
self._total_mem = long(memory) * 1024
self._total_mem = int(memory) * 1024
self._no_cpus = cpus
self.app.log.debug('QubesHost: no_cpus={} memory_total={}'.format(
@ -283,7 +279,7 @@ class QubesHost(object):
self._physinfo = self.app.xc.physinfo()
except AttributeError:
raise NotImplementedError('This function requires Xen hypervisor')
return long(self._physinfo['free_memory'])
return int(self._physinfo['free_memory'])
def measure_cpu_usage(self, previous_time=None, previous=None,
@ -329,7 +325,7 @@ class QubesHost(object):
current[vm['domid']]['cpu_usage'] = (
float(current[vm['domid']]['cpu_time'] -
previous[vm['domid']]['cpu_time']) /
long(1000 ** 3) / (current_time - previous_time) * 100)
1000 ** 3 / (current_time - previous_time) * 100)
if current[vm['domid']]['cpu_usage'] < 0:
# VM has been rebooted
current[vm['domid']]['cpu_usage'] = 0
@ -428,7 +424,7 @@ class VMCollection(object):
if isinstance(key, int):
return self._dict[key]
if isinstance(key, basestring):
if isinstance(key, str):
for vm in self:
if vm.name == key:
return vm
@ -682,7 +678,7 @@ class Qubes(qubes.PropertyHolder):
try:
self.pools[name] = self._get_pool(**node.attrib)
except qubes.exc.QubesException as e:
self.log.error(e.message)
self.log.error(str(e))
# stage 2: load VMs
for node in self.xml.xpath('./domains/domain'):
@ -783,7 +779,7 @@ class Qubes(qubes.PropertyHolder):
lxml.etree.ElementTree(self.__xml__()).write(
fh_new, encoding='utf-8', pretty_print=True)
fh_new.flush()
os.chmod(fh_new.name, 0660)
os.chmod(fh_new.name, 0o660)
os.chown(fh_new.name, -1, grp.getgrnam('qubes').gr_gid)
os.rename(fh_new.name, self._store)

View File

@ -1,5 +1,3 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -962,7 +960,7 @@ class ExtractWorker2(Process):
pass
process.wait()
self.log.error("ERROR: " + unicode(e))
raise e, None, exc_traceback
raise
def handle_dir_relocations(self, dirname):
''' Relocate files in given director when it's already extracted

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -47,6 +44,7 @@ Such extension should provide:
- handle `device-list-attached:class` event - list currently attached
devices to this domain
'''
import qubes.utils

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -29,11 +26,12 @@ particularly our custom Sphinx extension.
'''
import argparse
import io
import json
import os
import re
import StringIO
import urllib2
import urllib.error
import urllib.request
import docutils
import docutils.nodes
@ -64,10 +62,10 @@ def fetch_ticket_info(app, number):
:param app: Sphinx app object
:param str number: number of the ticket, without #
:rtype: mapping
:raises: urllib2.HTTPError
:raises: urllib.error.HTTPError
'''
response = urllib2.urlopen(urllib2.Request(
response = urllib.request.urlopen(urllib.request.Request(
app.config.ticket_base_uri.format(number=number),
headers={
'Accept': 'application/vnd.github.v3+json',
@ -99,7 +97,7 @@ def ticket(name, rawtext, text, lineno, inliner, options=None, content=None):
try:
info = fetch_ticket_info(inliner.document.settings.env.app, ticketno)
except urllib2.HTTPError, e:
except urllib.error.HTTPError as e:
msg = inliner.reporter.error(
'Error while fetching ticket info: {!s}'.format(e), line=lineno)
prb = inliner.problematic(rawtext, rawtext, msg)
@ -167,7 +165,7 @@ def make_rst_section(heading, char):
def prepare_manpage(command):
parser = qubes.tools.get_parser_for_command(command)
stream = StringIO.StringIO()
stream = io.StringIO()
stream.write('.. program:: {}\n\n'.format(command))
stream.write(make_rst_section(
':program:`{}` -- {}'.format(command, parser.description), '='))

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -94,15 +91,13 @@ class EmitterMeta(type):
cls.add_handler(event, attr)
class Emitter(object):
class Emitter(object, metaclass=EmitterMeta):
'''Subject that can emit events.
By default all events are disabled not to interfere with loading from XML.
To enable event dispatch, set :py:attr:`events_enabled` to :py:obj:`True`.
'''
__metaclass__ = EmitterMeta
def __init__(self, *args, **kwargs):
super(Emitter, self).__init__(*args, **kwargs)
if not hasattr(self, 'events_enabled'):

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/env python
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -20,6 +18,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import os
import re
import subprocess

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -21,6 +19,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import datetime
import qubes.ext
import qubes.firewall

View File

@ -1,6 +1,5 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=too-few-public-methods
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -21,6 +20,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import datetime
import subprocess
@ -54,7 +54,7 @@ class RuleChoice(RuleOption):
super(RuleChoice, self).__init__(value)
self.allowed_values = \
[v for k, v in self.__class__.__dict__.items()
if not k.startswith('__') and isinstance(v, basestring) and
if not k.startswith('__') and isinstance(v, str) and
not v.startswith('__')]
if value not in self.allowed_values:
raise ValueError(value)
@ -457,7 +457,7 @@ class Firewall(object):
try:
old_umask = os.umask(0o002)
with open(firewall_conf, 'w') as firewall_xml:
with open(firewall_conf, 'wb') as firewall_xml:
xml_tree.write(firewall_xml, encoding="UTF-8",
pretty_print=True)
os.umask(old_umask)

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
@ -22,7 +20,6 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
#
import logging
import os

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
@ -22,7 +20,7 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
#
import logging
import string

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
@ -21,11 +19,11 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
#
import socket
import fcntl
class QMemmanClient:
class QMemmanClient:
def request_memory(self, amount):
self.sock = socket.socket(socket.AF_UNIX)

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -23,6 +20,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
""" Qubes storage system"""
from __future__ import absolute_import
@ -325,7 +323,7 @@ class Storage(object):
def resize(self, volume, size):
''' Resizes volume a read-writable volume '''
if isinstance(volume, basestring):
if isinstance(volume, str):
volume = self.vm.volumes[volume]
self.get_pool(volume).resize(volume, size)
if self.vm.is_running():
@ -444,7 +442,7 @@ class Storage(object):
def get_pool(self, volume):
''' Helper function '''
assert isinstance(volume, (Volume, basestring)), \
assert isinstance(volume, (Volume, str)), \
"You need to pass a Volume or pool name as str"
if isinstance(volume, Volume):
return self.pools[volume.name]
@ -474,7 +472,7 @@ class Storage(object):
def export(self, volume):
''' Helper function to export volume (pool.export(volume))'''
assert isinstance(volume, (Volume, basestring)), \
assert isinstance(volume, (Volume, str)), \
"You need to pass a Volume or pool name as str"
if isinstance(volume, Volume):
return self.pools[volume.name].export(volume)

View File

@ -1,5 +1,3 @@
#!/usr/bin/env python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,8 +17,10 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Manages block devices in a domain '''
import itertools
import string # pylint: disable=deprecated-module
from qubes.storage import Pool, Volume
@ -43,9 +43,10 @@ class DomainPool(Pool):
''' Queries qubesdb and returns volumes for `self.vm` '''
qdb = self.vm.qdb
safe_set = set(string.letters + string.digits + string.punctuation)
safe_set = set(itertools.chain(
string.ascii_letters, string.digits, string.punctuation))
allowed_attributes = {'desc': string.printable,
'mode': string.letters,
'mode': string.ascii_letters,
'size': string.digits}
if not self.vm.is_running():
return []

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,6 +20,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' This module contains pool implementations backed by file images'''
from __future__ import absolute_import
@ -67,7 +66,7 @@ class FilePool(qubes.storage.Pool):
return target
def create(self, volume):
assert isinstance(volume.size, (int, long)) and volume.size > 0, \
assert isinstance(volume.size, int) and volume.size > 0, \
'Volatile volume size must be > 0'
if volume._is_origin:
create_sparse_file(volume.path, volume.size)
@ -157,7 +156,7 @@ class FilePool(qubes.storage.Pool):
# TODO: Renaming the old revisions
new_path = os.path.join(self.dir_path, subdir, new_name)
if not os.path.exists(new_path):
os.mkdir(new_path, 0755)
os.mkdir(new_path, 0o755)
new_volume_path = os.path.join(new_path, self.name + '.img')
if not volume.backward_comp:
os.rename(volume.path, new_volume_path)
@ -203,7 +202,7 @@ class FilePool(qubes.storage.Pool):
def reset(self, volume):
''' Remove and recreate a volatile volume '''
assert volume._is_volatile, "Not a volatile volume"
assert isinstance(volume.size, (int, long)) and volume.size > 0, \
assert isinstance(volume.size, int) and volume.size > 0, \
'Volatile volume size must be > 0'
_remove_if_exists(volume.path)

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -21,7 +19,9 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' This module contains pool implementations for different OS kernels. '''
import os
from qubes.storage import Pool, StoragePoolException, Volume

View File

@ -1,5 +1,3 @@
# vim: fileencoding=utf-8
# pylint: disable=abstract-method
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,6 +17,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Driver for storing vm images in a LVM thin pool '''
import logging

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -27,7 +25,6 @@ import io
BUF_SIZE = 409600
class TarSparseInfo(tarfile.TarInfo):
def __init__(self, name="", sparsemap=None):
super(TarSparseInfo, self).__init__(name)
@ -48,12 +45,12 @@ class TarSparseInfo(tarfile.TarInfo):
def sparse_header_chunk(self, index):
if index < len(self.sparsemap):
return ''.join([
return b''.join([
tarfile.itn(self.sparsemap[index][0], 12, tarfile.GNU_FORMAT),
tarfile.itn(self.sparsemap[index][1], 12, tarfile.GNU_FORMAT),
])
else:
return '\0' * 12 * 2
return b'\0' * 12 * 2
def get_gnu_header(self):
'''Part placed in 'prefix' field of posix header'''
@ -62,20 +59,20 @@ class TarSparseInfo(tarfile.TarInfo):
tarfile.itn(self.mtime, 12, tarfile.GNU_FORMAT), # atime
tarfile.itn(self.mtime, 12, tarfile.GNU_FORMAT), # ctime
tarfile.itn(0, 12, tarfile.GNU_FORMAT), # offset
tarfile.stn('', 4), # longnames
'\0', # unused_pad2
tarfile.stn('', 4, tarfile.ENCODING, 'surrogateescape'), #longnames
b'\0', # unused_pad2
]
parts += [self.sparse_header_chunk(i) for i in range(4)]
parts += [
'\1' if len(self.sparsemap) > 4 else '\0', # isextended
b'\1' if len(self.sparsemap) > 4 else b'\0', # isextended
tarfile.itn(self.realsize, 12, tarfile.GNU_FORMAT), # realsize
]
return ''.join(parts)
return b''.join(parts)
def get_info(self, encoding, errors):
info = super(TarSparseInfo, self).get_info(encoding, errors)
def get_info(self):
info = super(TarSparseInfo, self).get_info()
# place GNU extension into
info['prefix'] = self.get_gnu_header()
info['prefix'] = self.get_gnu_header().decode(tarfile.ENCODING)
return info
def tobuf(self, format=tarfile.DEFAULT_FORMAT, encoding=tarfile.ENCODING,
@ -83,16 +80,20 @@ class TarSparseInfo(tarfile.TarInfo):
# pylint: disable=redefined-builtin
header_buf = super(TarSparseInfo, self).tobuf(format, encoding, errors)
if len(self.sparsemap) > 4:
return header_buf + ''.join(self.create_ext_sparse_headers())
return header_buf + b''.join(self.create_ext_sparse_headers())
else:
return header_buf
def create_ext_sparse_headers(self):
for ext_hdr in range(4, len(self.sparsemap), 21):
sparse_parts = [self.sparse_header_chunk(i) for i in
range(ext_hdr, ext_hdr+21)]
sparse_parts += '\1' if ext_hdr+21 < len(self.sparsemap) else '\0'
yield tarfile.stn(''.join(sparse_parts), 512)
sparse_parts = [
self.sparse_header_chunk(i).decode(
tarfile.ENCODING, 'surrogateescape')
for i in range(ext_hdr, ext_hdr+21)]
sparse_parts.append(
'\1' if ext_hdr+21 < len(self.sparsemap) else '\0')
yield tarfile.stn(''.join(sparse_parts), 512,
tarfile.ENCODING, 'surrogateescape')
def get_sparse_map(input_file):
@ -161,8 +162,8 @@ def copy_sparse_data(input_stream, output_stream, sparse_map):
def finalize(output):
'''Write EOF blocks'''
output.write('\0' * 512)
output.write('\0' * 512)
output.write(b'\0' * 512)
output.write(b'\0' * 512)
def main(args=None):
parser = argparse.ArgumentParser()

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=invalid-name
#
@ -33,7 +31,6 @@
don't run the tests.
"""
import __builtin__
import collections
import functools
import logging
@ -90,7 +87,7 @@ except libvirt.libvirtError:
try:
in_git = subprocess.check_output(
['git', 'rev-parse', '--show-toplevel']).strip()
['git', 'rev-parse', '--show-toplevel']).decode().strip()
qubes.log.LOGPATH = '/tmp'
qubes.log.LOGFILE = '/tmp/qubes.log'
except subprocess.CalledProcessError:
@ -166,7 +163,7 @@ def expectedFailureIfTemplate(templates):
@functools.wraps(func)
def wrapper(self, *args, **kwargs):
template = self.template
if isinstance(templates, basestring):
if isinstance(templates, str):
should_expect_fail = template in templates
else:
should_expect_fail = any([template in x for x in templates])
@ -283,6 +280,10 @@ class QubesTestCase(unittest.TestCase):
def tearDown(self):
super(QubesTestCase, self).tearDown()
# TODO: find better way in py3
try:
result = self._outcome.result
except:
result = self._resultForDoCleanups
failed_test_cases = result.failures \
+ result.errors \
@ -334,7 +335,7 @@ class QubesTestCase(unittest.TestCase):
self.assertEqual(xml1.tag, xml2.tag)
self.assertEqual(xml1.text, xml2.text)
self.assertItemsEqual(xml1.keys(), xml2.keys())
self.assertCountEqual(xml1.keys(), xml2.keys())
for key in xml1.keys():
self.assertEqual(xml1.get(key), xml2.get(key))
@ -503,7 +504,7 @@ class SystemTestsMixin(object):
def init_default_template(self, template=None):
if template is None:
template = self.host_app.default_template
elif isinstance(template, basestring):
elif isinstance(template, str):
template = self.host_app.domains[template]
template_vm = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM,
@ -752,9 +753,8 @@ class SystemTestsMixin(object):
wait_count = 0
while subprocess.call(['xdotool', 'search', '--name', title],
stdout=open(os.path.devnull, 'w'),
stderr=subprocess.STDOUT) == \
__builtin__.int(show):
stdout=open(os.path.devnull, 'w'), stderr=subprocess.STDOUT) \
== int(show):
wait_count += 1
if wait_count > timeout*10:
self.fail("Timeout while waiting for {} window to {}".format(
@ -841,7 +841,7 @@ class SystemTestsMixin(object):
init_path = os.path.join(mountpoint, 'init')
with open(init_path, 'w') as f:
f.write(init_script)
os.chmod(init_path, 0755)
os.chmod(init_path, 0o755)
dracut_args = [
'--kver', kernel_version,
'--include', init_path,
@ -894,23 +894,23 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
for modname in (
# integration tests
'qubes.tests.int.basic',
'qubes.tests.int.storage',
'qubes.tests.int.devices_pci',
'qubes.tests.int.dom0_update',
'qubes.tests.int.network',
'qubes.tests.int.dispvm',
'qubes.tests.int.vm_qrexec_gui',
'qubes.tests.int.backup',
'qubes.tests.int.backupcompatibility',
'qubes.tests.integ.basic',
'qubes.tests.integ.storage',
'qubes.tests.integ.devices_pci',
'qubes.tests.integ.dom0_update',
'qubes.tests.integ.network',
'qubes.tests.integ.dispvm',
'qubes.tests.integ.vm_qrexec_gui',
'qubes.tests.integ.backup',
'qubes.tests.integ.backupcompatibility',
# 'qubes.tests.regressions',
# tool tests
'qubes.tests.int.tools.qubes_create',
'qubes.tests.int.tools.qvm_check',
'qubes.tests.int.tools.qvm_firewall',
'qubes.tests.int.tools.qvm_prefs',
'qubes.tests.int.tools.qvm_run',
'qubes.tests.integ.tools.qubes_create',
'qubes.tests.integ.tools.qvm_check',
'qubes.tests.integ.tools.qvm_firewall',
'qubes.tests.integ.tools.qvm_prefs',
'qubes.tests.integ.tools.qvm_run',
# external modules
# 'qubes.tests.extra',

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#
@ -32,26 +30,7 @@ import qubes
import qubes.events
import qubes.tests
# FIXME: blatant duplication with qubes.tests.init
class TestVM(qubes.vm.BaseVM):
qid = qubes.property('qid', type=int)
name = qubes.property('name')
netid = qid
uuid = uuid.uuid5(uuid.NAMESPACE_DNS, 'testvm')
class MockLibvirt(object):
def undefine(self):
pass
libvirt_domain = MockLibvirt()
def is_halted(self):
return True
def get_power_state(self):
return "Halted"
import qubes.tests.init
class TestApp(qubes.tests.TestEmitter):
pass
@ -61,8 +40,10 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
self.app = TestApp()
self.vms = qubes.app.VMCollection(self.app)
self.testvm1 = TestVM(None, None, qid=1, name='testvm1')
self.testvm2 = TestVM(None, None, qid=2, name='testvm2')
self.testvm1 = qubes.tests.init.TestVM(
None, None, qid=1, name='testvm1')
self.testvm2 = qubes.tests.init.TestVM(
None, None, qid=2, name='testvm2')
def test_000_contains(self):
self.vms._dict = {1: self.testvm1}
@ -91,8 +72,10 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
with self.assertRaises(TypeError):
self.vms.add(object())
testvm_qid_collision = TestVM(None, None, name='testvm2', qid=1)
testvm_name_collision = TestVM(None, None, name='testvm1', qid=2)
testvm_qid_collision = qubes.tests.init.TestVM(
None, None, name='testvm2', qid=1)
testvm_name_collision = qubes.tests.init.TestVM(
None, None, name='testvm1', qid=2)
with self.assertRaises(ValueError):
self.vms.add(testvm_qid_collision)
@ -103,27 +86,27 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.qids(), [1, 2])
self.assertItemsEqual(self.vms.keys(), [1, 2])
self.assertCountEqual(self.vms.qids(), [1, 2])
self.assertCountEqual(self.vms.keys(), [1, 2])
def test_004_names(self):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.names(), ['testvm1', 'testvm2'])
self.assertCountEqual(self.vms.names(), ['testvm1', 'testvm2'])
def test_005_vms(self):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.vms(), [self.testvm1, self.testvm2])
self.assertItemsEqual(self.vms.values(), [self.testvm1, self.testvm2])
self.assertCountEqual(self.vms.vms(), [self.testvm1, self.testvm2])
self.assertCountEqual(self.vms.values(), [self.testvm1, self.testvm2])
def test_006_items(self):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.items(),
self.assertCountEqual(self.vms.items(),
[(1, self.testvm1), (2, self.testvm2)])
def test_007_len(self):
@ -138,7 +121,7 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
del self.vms['testvm2']
self.assertItemsEqual(self.vms.vms(), [self.testvm1])
self.assertCountEqual(self.vms.vms(), [self.testvm1])
self.assertEventFired(self.app, 'domain-delete', args=[self.testvm2])
def test_100_get_new_unused_qid(self):

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -80,7 +77,7 @@ class TC_00_Emitter(qubes.tests.QubesTestCase):
effect = emitter.fire_event('testevent')
self.assertItemsEqual(effect,
self.assertCountEqual(effect,
('testvalue1', 'testvalue2', 'testvalue3', 'testvalue4'))
def test_004_catch_all(self):

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -76,9 +73,8 @@ def load_tests(loader, tests, pattern):
for test_case in entry.load()():
tests.addTests(loader.loadTestsFromTestCase(test_case))
except Exception as err: # pylint: disable=broad-except
tb = sys.exc_info()[2]
def runTest(self):
raise err, None, tb
raise err
ExtraLoadFailure = type('ExtraLoadFailure',
(qubes.tests.QubesTestCase,),
{entry.name: runTest})
@ -105,9 +101,8 @@ def load_tests(loader, tests, pattern):
)
))
except Exception as err: # pylint: disable=broad-except
tb = sys.exc_info()[2]
def runTest(self):
raise err, None, tb
raise err
ExtraForTemplateLoadFailure = type('ExtraForTemplateLoadFailure',
(qubes.tests.QubesTestCase,),
{entry.name: runTest})

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#
@ -292,6 +290,12 @@ class TestVM(qubes.vm.BaseVM):
netid = qid
uuid = uuid.uuid5(uuid.NAMESPACE_DNS, 'testvm')
def __lt__(self, other):
try:
return self.name < other.name
except AttributeError:
return NotImplemented
class MockLibvirt(object):
def undefine(self):
pass
@ -304,6 +308,7 @@ class TestVM(qubes.vm.BaseVM):
def get_power_state(self):
return "Halted"
class TestApp(qubes.tests.TestEmitter):
pass
@ -354,27 +359,27 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.qids(), [1, 2])
self.assertItemsEqual(self.vms.keys(), [1, 2])
self.assertCountEqual(self.vms.qids(), [1, 2])
self.assertCountEqual(self.vms.keys(), [1, 2])
def test_004_names(self):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.names(), ['testvm1', 'testvm2'])
self.assertCountEqual(self.vms.names(), ['testvm1', 'testvm2'])
def test_005_vms(self):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.vms(), [self.testvm1, self.testvm2])
self.assertItemsEqual(self.vms.values(), [self.testvm1, self.testvm2])
self.assertCountEqual(self.vms.vms(), [self.testvm1, self.testvm2])
self.assertCountEqual(self.vms.values(), [self.testvm1, self.testvm2])
def test_006_items(self):
self.vms.add(self.testvm1)
self.vms.add(self.testvm2)
self.assertItemsEqual(self.vms.items(),
self.assertCountEqual(self.vms.items(),
[(1, self.testvm1), (2, self.testvm2)])
def test_007_len(self):
@ -389,7 +394,7 @@ class TC_30_VMCollection(qubes.tests.QubesTestCase):
del self.vms['testvm2']
self.assertItemsEqual(self.vms.vms(), [self.testvm1])
self.assertCountEqual(self.vms.vms(), [self.testvm1])
self.assertEventFired(self.app, 'domain-delete', args=[self.testvm2])
def test_100_get_new_unused_qid(self):

View File

@ -1,6 +1,3 @@
#!/usr/bin/python
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,6 +19,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import hashlib
import logging
import multiprocessing

View File

@ -1,5 +1,3 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,7 +17,7 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
#
from multiprocessing import Queue
import os
@ -31,7 +29,7 @@ import sys
import re
import qubes.tests
import qubes.tests.int.backup
import qubes.tests.integ.backup
QUBESXML_R2B2 = '''
<QubesVmCollection updatevm="3" default_kernel="3.7.6-2" default_netvm="3" default_fw_netvm="2" default_template="1" clockvm="2">
@ -145,7 +143,7 @@ compression-filter=gzip
'''
class TC_00_BackupCompatibility(
qubes.tests.int.backup.BackupTestsMixin, qubes.tests.QubesTestCase):
qubes.tests.integ.backup.BackupTestsMixin, qubes.tests.QubesTestCase):
def tearDown(self):
self.remove_test_vms(prefix="test-")

View File

@ -1,6 +1,5 @@
#!/usr/bin/python
# vim: fileencoding=utf-8
# pylint: disable=invalid-name
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,6 +21,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
from distutils import spawn
import os

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#
@ -22,6 +20,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import os
import subprocess
import time

View File

@ -1,5 +1,3 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -20,7 +18,6 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
#
from distutils import spawn
import qubes.tests

View File

@ -1,5 +1,3 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -20,6 +18,7 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.
#
import os
import shutil
import subprocess

View File

@ -1,6 +1,3 @@
#!/usr/bin/python
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,6 +19,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
from distutils import spawn
import multiprocessing

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,6 +17,8 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import os
import shutil

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,7 +19,6 @@
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes
import qubes.tools.qubes_create

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,6 +17,7 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes.firewall
import qubes.tests

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,7 +19,6 @@
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes
import qubes.tools.qvm_prefs

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -22,6 +19,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
from distutils import spawn
import multiprocessing

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -24,6 +21,7 @@
import argparse
import curses
import itertools
import logging
import logging.handlers
import os
@ -36,30 +34,38 @@ import unittest.signals
import qubes.tests
class CursesColor(dict):
colors = (
'black', 'red', 'green', 'yellow', 'blue', 'magenta', 'cyan', 'white')
attrs = {
'bold': 'bold', 'normal': 'sgr0'}
def __init__(self):
super(CursesColor, self).__init__()
self.has_colors = False
try:
curses.setupterm()
self.has_colors = True
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)
self['yellow'] = curses.tparm(curses.tigetstr('setaf'), 3)
self['blue'] = curses.tparm(curses.tigetstr('setaf'), 4)
self['magenta'] = curses.tparm(curses.tigetstr('setaf'), 5)
self['cyan'] = curses.tparm(curses.tigetstr('setaf'), 6)
self['white'] = curses.tparm(curses.tigetstr('setaf'), 7)
self['bold'] = curses.tigetstr('bold')
self['normal'] = curses.tigetstr('sgr0')
def __missing__(self, key):
# pylint: disable=unused-argument,no-self-use
if not self.has_colors:
return ''
try:
value = curses.tigetstr(self.attrs[key])
except KeyError:
try:
value = curses.tparm(
curses.tigetstr('setaf'), self.colors.index(key))
except ValueError:
return ''
value = value.decode()
self[key] = value
return value
class QubesTestResult(unittest.TestResult):
'''A test result class that can print colourful text results to a stream.
@ -293,13 +299,18 @@ parser.add_argument('--do-clean', '-C',
help='do execute tearDown even on failed tests.')
# pylint: disable=protected-access
try:
name_to_level = logging._nameToLevel
except AttributeError:
name_to_level = logging._levelNames
parser.add_argument('--loglevel', '-L', metavar='LEVEL',
action='store', choices=tuple(k
for k in sorted(logging._levelNames.keys(),
key=lambda x: logging._levelNames[x])
if isinstance(k, basestring)),
for k in sorted(name_to_level.keys(),
key=lambda x: name_to_level[x])
if isinstance(k, str)),
help='logging level for file and syslog forwarding '
'(one of: %(choices)s; default: %(default)s)')
del name_to_level
# pylint: enable=protected-access
parser.add_argument('--logfile', '-o', metavar='FILE',

View File

@ -1,3 +1,4 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
@ -15,6 +16,7 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes.log
from qubes.exc import QubesException
@ -80,7 +82,7 @@ class TC_00_Pool(QubesTestCase):
def test_001_all_pool_drivers(self):
""" The only predefined pool driver is file """
self.assertEquals(['linux-kernel', 'lvm_thin', 'file'], pool_drivers())
self.assertCountEqual(['linux-kernel', 'lvm_thin', 'file'], pool_drivers())
def test_002_get_pool_klass(self):
""" Expect the default pool to be `FilePool` """

View File

@ -1,3 +1,4 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
@ -15,6 +16,7 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Tests for the file storage backend '''

View File

@ -1,5 +1,3 @@
# vim: fileencoding=utf-8
# pylint: disable=missing-docstring
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,6 +17,8 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import os
import subprocess
import tempfile
@ -58,7 +58,7 @@ class TC_00_TarWriter(qubes.tests.QubesTestCase):
expected_output = (
'tar: Removing leading `/\' from member names\n' +
expected_output)
self.assertEqual(tar_output, expected_output)
self.assertEqual(tar_output.decode(), expected_output)
extracted_path = os.path.join(self.extract_dir,
expected_name.lstrip('/'))
with self.assertNotRaises(subprocess.CalledProcessError):

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -20,18 +17,15 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import io
import sys
try:
import StringIO
except ImportError:
from io import StringIO
class StdoutBuffer(object):
def __init__(self):
self.stdout = StringIO.StringIO()
self.stdout = io.StringIO()
def __enter__(self):
sys.stdout = self.stdout

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -19,6 +17,8 @@
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import argparse
import qubes.firewall

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access,pointless-statement
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access
#
@ -109,8 +107,8 @@ class TC_10_BaseVM(qubes.tests.QubesTestCase):
'testfeature_aqq': 'aqq',
})
self.assertItemsEqual(vm.devices.keys(), ('pci',))
self.assertItemsEqual(list(vm.devices['pci'].attached(persistent=True)),
self.assertCountEqual(vm.devices.keys(), ('pci',))
self.assertCountEqual(list(vm.devices['pci'].attached(persistent=True)),
[qubes.ext.pci.PCIDevice(vm, '00:11.22')])
self.assertXMLIsValid(vm.__xml__(), 'domain.rng')

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access
#
@ -22,6 +20,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import unittest
import qubes

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
# pylint: disable=protected-access
#
@ -22,6 +20,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import os
import unittest
@ -244,7 +243,7 @@ class TC_90_QubesVM(QubesVMTestsMixin,qubes.tests.QubesTestCase):
def test_110_name(self):
vm = self.get_vm()
self.assertIsInstance(vm.name, basestring)
self.assertIsInstance(vm.name, str)
def test_120_uuid(self):
my_uuid = uuid.uuid4()

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -294,7 +291,7 @@ class PoolsAction(QubesAction):
pools = [app.get_pool(name) for name in pool_names]
setattr(namespace, self.dest, pools)
except qubes.exc.QubesException as e:
parser.error(e.message)
parser.error(str(e))
sys.exit(2)
@ -530,9 +527,9 @@ def print_table(table):
if sys.stdout != sys.__stdout__:
p = subprocess.Popen(cmd + ['-c', '80'], stdin=subprocess.PIPE,
stdout=subprocess.PIPE)
p.stdin.write(text_table)
p.stdin.write(text_table.encode())
(out, _) = p.communicate()
sys.stdout.write(out)
sys.stdout.write(out.decode())
else:
p = subprocess.Popen(cmd, stdin=subprocess.PIPE)
p.communicate(text_table)
p.communicate(text_table.encode())

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- coding: utf-8 -*-
# pylint: skip-file
#
@ -283,7 +281,7 @@ def main():
log.debug('instantiating server')
os.umask(0)
server = SocketServer.UnixStreamServer(SOCK_PATH, QMemmanReqHandler)
os.umask(077)
os.umask(0o077)
# notify systemd
nofity_socket = os.getenv('NOTIFY_SOCKET')

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -145,7 +143,7 @@ def main(args=None):
pass_f.close()
else:
if raw_input("Do you want to proceed? [y/N] ").upper() != "Y":
if input("Do you want to proceed? [y/N] ").upper() != "Y":
return 0
prompt = ("Please enter the passphrase that will be used to {}verify "

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -249,7 +247,7 @@ def main(args=None):
parser.error_runtime(str(e))
if args.pass_file is None:
if raw_input("Do you want to proceed? [y/N] ").upper() != "Y":
if input("Do you want to proceed? [y/N] ").upper() != "Y":
exit(0)
try:

View File

@ -1,6 +1,5 @@
#!/usr/bin/python2
# pylint: disable=C,R
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -132,10 +131,10 @@ def list_volumes(args):
if hasattr(args, 'domains') and args.domains:
result = [x # reduce to only VolumeData with assigned domains
for p in vd_dict.itervalues() for x in p.itervalues()
for p in vd_dict.values() for x in p.values()
if x.domains]
else:
result = [x for p in vd_dict.itervalues() for x in p.itervalues()]
result = [x for p in vd_dict.values() for x in p.values()]
qubes.tools.print_table(prepare_table(result, full=args.full))
def revert_volume(args):
@ -145,7 +144,7 @@ def revert_volume(args):
pool = app.pools[volume.pool]
pool.revert(volume)
except qubes.storage.StoragePoolException as e:
print(e.message, file=sys.stderr)
print(str(e), file=sys.stderr)
sys.exit(1)
def attach_volumes(args):
@ -158,7 +157,7 @@ def attach_volumes(args):
rw = not args.ro
vm.storage.attach(volume, rw=rw)
except qubes.storage.StoragePoolException as e:
print(e.message, file=sys.stderr)
print(str(e), file=sys.stderr)
sys.exit(1)
@ -171,7 +170,7 @@ def detach_volumes(args):
try:
vm.storage.detach(volume)
except qubes.storage.StoragePoolException as e:
print(e.message, file=sys.stderr)
print(str(e), file=sys.stderr)
sys.exit(1)
@ -266,7 +265,7 @@ def main(args=None):
args = parser.parse_args(args)
args.func(args)
except qubes.exc.QubesException as e:
parser.print_error(e.message)
parser.print_error(str(e))
return 1

View File

@ -1,6 +1,5 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
# :pylint: disable=too-few-public-methods
# pylint: disable=too-few-public-methods
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,6 +1,5 @@
#!/usr/bin/python2
# coding=utf-8
# pylint: disable=C,R
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -194,7 +193,7 @@ def main(args=None):
try:
args.func(args)
except qubes.exc.QubesException as e:
print(e.message, file=sys.stderr)
print(str(e), file=sys.stderr)
return 1
return 0

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -163,7 +161,7 @@ def main(args=None):
if args.reload:
vm.fire_event('firewall-changed')
except qubes.exc.QubesException as e:
parser.print_error(e.message)
parser.print_error(str(e))
return 1
return 0

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python -O
# vim: fileencoding=utf-8
# pylint: disable=too-few-public-methods
#
@ -28,7 +26,6 @@
from __future__ import print_function
import __builtin__
import argparse
import collections
import sys
@ -101,7 +98,7 @@ class Column(object):
ret = None
try:
if isinstance(self._attr, basestring):
if isinstance(self._attr, str):
ret = vm
for attrseg in self._attr.split('.'):
ret = getattr(ret, attrseg)
@ -151,7 +148,7 @@ def column(width=0, head=None):
def decorator(obj):
# pylint: disable=missing-docstring
# we keep hints on fget, so the order of decorators does not matter
holder = obj.fget if isinstance(obj, __builtin__.property) else obj
holder = obj.fget if isinstance(obj, property) else obj
try:
holder.ls_head = head or holder.__name__.replace('_', '-').upper()
@ -202,7 +199,7 @@ def process_class(cls):
for klass in cls.__mro__:
for prop in klass.__dict__.values():
holder = prop.fget \
if isinstance(prop, __builtin__.property) \
if isinstance(prop, property) \
else prop
if not hasattr(holder, 'ls_head') or holder.ls_head is None:
continue
@ -616,7 +613,7 @@ def main(args=None):
try:
args = parser.parse_args(args)
except qubes.exc.QubesException as e:
parser.print_error(e.message)
parser.print_error(str(e))
return 1
if args.fields:

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,5 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
# :pylint: disable=too-few-public-methods
# pylint: disable=too-few-public-methods
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -190,7 +189,7 @@ def main(args=None):
try:
args = parser.parse_args(args)
except qubes.exc.QubesException as e:
parser.print_error(e.message)
parser.print_error(str(e))
return 1
if args.command is None or args.command == 'list':

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/env python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2
# vim: fileencoding=utf8
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -143,7 +140,7 @@ class BaseVMMeta(qubes.events.EmitterMeta):
qubes.tools.qvm_ls.process_class(cls)
class BaseVM(qubes.PropertyHolder):
class BaseVM(qubes.PropertyHolder, metaclass=BaseVMMeta):
'''Base class for all VMs
:param app: Qubes application context
@ -157,8 +154,6 @@ class BaseVM(qubes.PropertyHolder):
'''
# pylint: disable=no-member
__metaclass__ = BaseVMMeta
def __init__(self, app, xml, features=None, devices=None, tags=None,
**kwargs):
# pylint: disable=redefined-outer-name

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, http://www.qubes-os.org
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -22,6 +20,7 @@
#
''' A disposable vm implementation '''
import copy
import qubes.vm.qubesvm

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -24,6 +22,7 @@
#
''' This module contains the NetVMMixin '''
import os
import re
@ -37,7 +36,7 @@ import qubes.exc
def _setter_mac(self, prop, value):
''' Helper for setting the MAC address '''
# pylint: disable=unused-argument
if not isinstance(value, basestring):
if not isinstance(value, str):
raise ValueError('MAC address must be a string')
value = value.lower()
if re.match(r"^([0-9a-f][0-9a-f]:){5}[0-9a-f][0-9a-f]$", value) is None:
@ -56,7 +55,7 @@ def _default_ip(self):
def _setter_ip(self, prop, value):
# pylint: disable=unused-argument
if not isinstance(value, basestring):
if not isinstance(value, str):
raise ValueError('IP address must be a string')
value = value.lower()
if re.match(r"^([0-9]{1,3}.){3}[0-9]{1,3}$", value) is None:

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#
@ -79,7 +77,7 @@ def _setter_qid(self, prop, value):
def _setter_name(self, prop, value):
''' Helper for setting the domain name '''
if not isinstance(value, basestring):
if not isinstance(value, str):
raise TypeError('{} value must be string, {!r} found'.format(
prop.__name__, type(value).__name__))
if len(value) > 31:
@ -135,7 +133,7 @@ def _setter_label(self, prop, value):
# pylint: disable=unused-argument
if isinstance(value, qubes.Label):
return value
if isinstance(value, basestring) and value.startswith('label-'):
if isinstance(value, str) and value.startswith('label-'):
return self.app.labels[int(value.split('-', 1)[1])]
return self.app.get_label(value)
@ -673,6 +671,9 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
def __hash__(self):
return self.qid
def __lt__(self, other):
return self.name < other.name
def __xml__(self):
element = super(QubesVM, self).__xml__()

View File

@ -1,6 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org/
#

View File

@ -1,5 +1,3 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
#
# The Qubes OS Project, http://www.qubes-os.org
#