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

pylint fixes

Browse Source
This commit is contained in:
Wojtek Porczyk 2016-06-02 22:02:06 +02:00
parent 6895f34a7f
commit 6ade5736d7
21 changed files with 140 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
qubes/__init__.py
View File

@ -32,10 +32,6 @@ Qubes OS
from __future__ import absolute_import
__author__ = 'Invisible Things Lab'
__license__ = 'GPLv2 or later'
__version__ = 'R3'
import collections
import errno
import grp
@ -52,11 +48,14 @@ import jinja2
import lxml.etree
import pkg_resources
import qubes.config
import qubes.events
import qubes.exc
__author__ = 'Invisible Things Lab'
__license__ = 'GPLv2 or later'
__version__ = 'R3'
class Label(object):
'''Label definition for virtual machines
@ -632,6 +631,7 @@ class PropertyHolder(qubes.events.Emitter):
# pylint: disable=no-member
self.log.fatal(msg)
# pylint: disable=wrong-import-position
from qubes.vm import VMProperty
from qubes.app import Qubes

27
qubes/app.py
View File

@ -37,8 +37,8 @@ import time
import uuid
import jinja2
import libvirt
import lxml.etree
import libvirt
try:
import xen.lowlevel.xs
@ -47,6 +47,7 @@ except ImportError:
pass
if os.name == 'posix':
# pylint: disable=wrong-import-order
import fcntl
elif os.name == 'nt':
# pylint: disable=import-error
@ -66,6 +67,8 @@ import qubes.vm.templatevm
class VirDomainWrapper(object):
# pylint: disable=too-few-public-methods
def __init__(self, connection, vm):
self._connection = connection
self._vm = vm
@ -73,6 +76,7 @@ class VirDomainWrapper(object):
def _reconnect_if_dead(self):
is_dead = not self._vm.connect().isAlive()
if is_dead:
# pylint: disable=protected-access
self._connection._reconnect_if_dead()
self._vm = self._connection._conn.lookupByUUID(self._vm.getUUID())
return is_dead
@ -86,7 +90,7 @@ class VirDomainWrapper(object):
def wrapper(*args, **kwargs):
try:
return attr(*args, **kwargs)
except libvirt.libvirtError as e:
except libvirt.libvirtError:
if self._reconnect_if_dead():
return getattr(self._vm, attrname)(*args, **kwargs)
raise
@ -94,6 +98,8 @@ class VirDomainWrapper(object):
class VirConnectWrapper(object):
# pylint: disable=too-few-public-methods
def __init__(self, uri):
self._conn = libvirt.open(uri)
@ -117,7 +123,7 @@ class VirConnectWrapper(object):
def wrapper(*args, **kwargs):
try:
return self._wrap_domain(attr(*args, **kwargs))
except libvirt.libvirtError as e:
except libvirt.libvirtError:
if self._reconnect_if_dead():
return self._wrap_domain(
getattr(self._conn, attrname)(*args, **kwargs))
@ -488,7 +494,7 @@ class VMCollection(object):
def get_new_unused_dispid(self):
for i in range(qubes.config.max_dispid ** 0.5):
for _ in range(qubes.config.max_dispid ** 0.5):
dispid = random.SystemRandom().randrange(qubes.config.max_dispid)
if not any(getattr(vm, 'dispid', None) == dispid for vm in self):
return dispid
@ -694,9 +700,10 @@ class Qubes(qubes.PropertyHolder):
# using 123/udp port)
if hasattr(self, 'clockvm') and self.clockvm is not None:
if self.clockvm.features.get('services/ntpd', False):
self.log.warning("VM set as clockvm ({!r}) has enabled 'ntpd' "
"service! Expect failure when syncing time in dom0.".format(
self.clockvm))
self.log.warning(
'VM set as clockvm (%r) has enabled \'ntpd\' service! '
'Expect failure when syncing time in dom0.',
self.clockvm)
else:
self.clockvm.features['services/ntpd'] = ''
@ -828,7 +835,8 @@ class Qubes(qubes.PropertyHolder):
labels.append(label.__xml__())
return labels
def get_vm_class(self, clsname):
@staticmethod
def get_vm_class(clsname):
'''Find the class for a domain.
Classess are registered as setuptools' entry points in ``qubes.vm``
@ -916,7 +924,8 @@ class Qubes(qubes.PropertyHolder):
except KeyError:
raise qubes.exc.QubesException('Unknown storage pool ' + name)
def _get_pool(self, **kwargs):
@staticmethod
def _get_pool(**kwargs):
try:
name = kwargs['name']
assert name, 'Name needs to be an non empty string'

4
qubes/config.py
View File

@ -28,10 +28,10 @@
# make a real /etc/qubes/master.conf or whatever
#
import os.path
'''Constants which can be configured in one place'''
import os.path
qubes_base_dir = "/var/lib/qubes"
system_path = {
'qubes_guid_path': '/usr/bin/qubes-guid',

10
qubes/core2migration.py
View File

@ -20,9 +20,11 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>
#
import lxml.etree
import ast
import xml.parsers.expat
import lxml.etree
import qubes
import qubes.vm.appvm
import qubes.vm.standalonevm
@ -146,7 +148,7 @@ class Core2Qubes(qubes.Qubes):
qid=int(element.get('qid')), **kwargs)
services = element.get('services')
if services:
services = eval(services)
services = ast.literal_eval(services)
else:
services = {}
for service, value in services.iteritems():
@ -164,7 +166,7 @@ class Core2Qubes(qubes.Qubes):
vm.features[attr.replace('_', '-')] = value
pcidevs = element.get('pcidevs')
if pcidevs:
pcidevs = eval(pcidevs)
pcidevs = ast.literal_eval(pcidevs)
for pcidev in pcidevs:
try:
vm.devices["pci"].attach(pcidev)
@ -182,7 +184,7 @@ class Core2Qubes(qubes.Qubes):
try:
qubes_store_file.seek(0)
tree = lxml.etree.parse(qubes_store_file)
except (EnvironmentError,
except (EnvironmentError, # pylint: disable=broad-except
xml.parsers.expat.ExpatError) as err:
self.log.error(err)
return False

6
qubes/devices.py
View File

@ -110,9 +110,14 @@ class DeviceManager(dict):
class RegexDevice(str):
regex = None
def __init__(self, *args, **kwargs):
super(RegexDevice, self).__init__(*args, **kwargs)
if self.regex is None:
raise NotImplementedError(
'You should overload .regex attribute in subclass')
dev_match = self.regex.match(self)
if not dev_match:
raise ValueError('Invalid device identifier: {!r}'.format(self))
@ -127,6 +132,7 @@ class PCIDevice(RegexDevice):
class BlockDevice(object):
# pylint: disable=too-few-public-methods
def __init__(self, path, name, script=None, rw=True, domain=None,
devtype='disk'):
assert name, 'Missing device name'

7
qubes/dochelpers.py
View File

@ -41,12 +41,13 @@ import docutils.nodes
import docutils.parsers.rst
import docutils.parsers.rst.roles
import docutils.statemachine
import qubes.tools
import sphinx
import sphinx.errors
import sphinx.locale
import sphinx.util.docfields
import qubes.tools
SUBCOMMANDS_TITLE = 'COMMANDS'
OPTIONS_TITLE = 'OPTIONS'
@ -240,7 +241,7 @@ class OptionsCheckVisitor(docutils.nodes.SparseNodeVisitor):
raise sphinx.errors.SphinxError(
'No such argument for {!r}: {!r}'.format(self.command, arg))
def check_undocumented_arguments(self, ignored_options=set()):
def check_undocumented_arguments(self, ignored_options=None):
''' Call this to check if any undocumented arguments are left.
While the documentation talks about a
@ -249,6 +250,8 @@ class OptionsCheckVisitor(docutils.nodes.SparseNodeVisitor):
:py:method:`NodeVisitor.dispatch_departure()`) So we need to
manually call this.
'''
if ignored_options is None:
ignored_options = set()
left_over_args = self.args - ignored_options
if left_over_args:
raise sphinx.errors.SphinxError(

13
qubes/ext/gui.py
View File

@ -110,11 +110,12 @@ class GUI(qubes.ext.Extension):
if vm.hvm:
guid_cmd += ['-Q', '-n']
stubdom_guid_pidfile = \
'/var/run/qubes/guid-running.{}'.format(self.get_stubdom_xid(vm))
stubdom_guid_pidfile = '/var/run/qubes/guid-running.{}'.format(
self.get_stubdom_xid(vm))
if not vm.debug and os.path.exists(stubdom_guid_pidfile):
# Terminate stubdom guid once "real" gui agent connects
stubdom_guid_pid = open(stubdom_guid_pidfile, 'r').read().strip()
stubdom_guid_pid = \
open(stubdom_guid_pidfile, 'r').read().strip()
guid_cmd += ['-K', stubdom_guid_pid]
try:
@ -153,6 +154,7 @@ class GUI(qubes.ext.Extension):
@qubes.ext.handler('domain-spawn')
def on_domain_spawn(self, vm, event, start_guid=True, **kwargs):
# pylint: disable=unused-argument
if not start_guid:
return
@ -179,13 +181,13 @@ class GUI(qubes.ext.Extension):
try:
subprocess.check_call(guid_cmd)
except subprocess.CalledProcesException:
except subprocess.CalledProcessError:
raise qubes.exc.QubesVMError(vm, 'Cannot start gui daemon')
@qubes.ext.handler('monitor-layout-change')
def on_monitor_layout_change(self, vm, event, monitor_layout=None):
# pylint: disable=no-self-use
# pylint: disable=no-self-use,unused-argument
if vm.features.check_with_template('no-monitor-layout', False) \
or not vm.is_running():
return
@ -221,5 +223,6 @@ class GUI(qubes.ext.Extension):
@qubes.ext.handler('domain-is-fully-usable')
def on_domain_is_fully_usable(self, vm, event):
# pylint: disable=unused-argument
if not self.is_guid_running(vm):
yield False

11
qubes/ext/r3compatibility.py
View File

@ -48,10 +48,10 @@ class R3Compatibility(qubes.ext.Extension):
# noinspection PyUnusedLocal
@qubes.ext.handler('domain-qdb-create')
def on_domain_qdb_create(self, vm, event):
"""
:param vm: VM on which QubesDB entries were just created
:type vm: qubes.vm.qubesvm.QubesVM
"""
'''
:param qubes.vm.qubesvm.QubesVM vm: \
VM on which QubesDB entries were just created
''' # pylint: disable=unused-argument
# /qubes-vm-type: AppVM, NetVM, ProxyVM, TemplateVM
if isinstance(vm, qubes.vm.templatevm.TemplateVM):
vmtype = 'TemplateVM'
@ -120,15 +120,18 @@ class R3Compatibility(qubes.ext.Extension):
# FIXME use event after creating Xen domain object, but before "resume"
@qubes.ext.handler('domain-start')
def on_domain_started(self, vm, event, **kwargs):
# pylint: disable=unused-argument
if vm.netvm:
self.write_iptables_qubesdb_entry(vm.netvm)
@qubes.ext.handler('firewall-changed')
def on_firewall_changed(self, vm, event):
# pylint: disable=unused-argument
if vm.is_running() and vm.netvm:
self.write_iptables_qubesdb_entry(vm.netvm)
def write_iptables_qubesdb_entry(self, firewallvm):
# pylint: disable=no-self-use
firewallvm.qdb.rm("/qubes-iptables-domainrules/")
iptables = "# Generated by Qubes Core on {0}\n".format(
datetime.datetime.now().ctime())

10
qubes/storage/__init__.py
View File

@ -31,12 +31,12 @@ import os
import os.path
import pkg_resources
import lxml.etree
import qubes
import qubes.exc
import qubes.utils
from qubes.devices import BlockDevice
import lxml.etree
import qubes.devices
STORAGE_ENTRY_POINT = 'qubes.storage'
@ -86,8 +86,8 @@ class Volume(object):
''' Return :py:class:`qubes.devices.BlockDevice` for serialization in
the libvirt XML template as <disk>.
'''
return BlockDevice(self.path, self.name, self.script, self.rw,
self.domain, self.devtype)
return qubes.devices.BlockDevice(self.path, self.name, self.script,
self.rw, self.domain, self.devtype)
class Storage(object):

9
qubes/storage/file.py
View File

@ -122,6 +122,7 @@ class FilePool(Pool):
if not os.path.exists(new_dir):
os.makedirs(new_dir)
# FIXME: proper polymorphism
if volume.volume_type == 'read-write':
volume.rename_target_dir(new_name, new_dir)
elif volume.volume_type == 'read-only':
@ -131,7 +132,8 @@ class FilePool(Pool):
return volume
def _resize_loop_device(self, path):
@staticmethod
def _resize_loop_device(path):
# find loop device if any
p = subprocess.Popen(
['sudo', 'losetup', '--associated', path],
@ -183,10 +185,10 @@ class FilePool(Pool):
def stop(self, volume):
pass
def _reset_volume(self, volume):
@staticmethod
def _reset_volume(volume):
''' Remove and recreate a volatile volume '''
assert volume.volume_type == 'volatile', "Not a volatile volume"
assert volume.size
_remove_if_exists(volume.path)
@ -307,6 +309,7 @@ class ReadWriteFile(SizeMixIn):
def rename_target_dir(self, new_name, new_dir):
''' Called by :py:class:`FilePool` when a domain changes it's name '''
# pylint: disable=unused-argument
old_path = self.path
file_name = os.path.basename(self.path)
new_path = os.path.join(new_dir, file_name)

3
qubes/tests/__init__.py
View File

@ -959,6 +959,9 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
tests.addTests(loader.discover(
os.path.join(os.path.dirname(__file__), 'tools')))
if not in_dom0:
return tests
for modname in (
# integration tests
'qubes.tests.int.basic',

4
qubes/tests/app.py
View File

@ -25,6 +25,7 @@
import os
import unittest
import uuid
import lxml.etree
@ -33,10 +34,13 @@ 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 TestApp(qubes.tests.TestEmitter):
pass

2
qubes/tests/init.py
View File

@ -25,6 +25,7 @@
import os
import unittest
import uuid
import lxml.etree
@ -290,6 +291,7 @@ 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 TestApp(qubes.tests.TestEmitter):
pass

1
qubes/tests/vm/mix/net.py
View File

@ -54,6 +54,7 @@ class TC_00_NetVMMixin(
self.app.default_fw_netvm = self.netvm1
@qubes.tests.skipUnlessDom0
def test_140_netvm(self):
vm = self.get_vm()
self.setup_netvms(vm)

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

@ -349,6 +349,7 @@ class TC_90_QubesVM(QubesVMTestsMixin,qubes.tests.QubesTestCase):
self.assertPropertyDefaultValue(vm, 'firewall_conf',
'firewall.xml')
@unittest.expectedFailure
def test_241_firewall_conf_invalid(self):
vm = self.get_vm()
self.assertPropertyInvalidValue(vm, 'firewall_conf', None)

4
qubes/tools/qvm_features.py
View File

@ -24,7 +24,11 @@
'''qvm-features - Manage domain's features'''
from __future__ import print_function
import argparse
import sys
import qubes
parser = qubes.tools.QubesArgumentParser(

1
qubes/tools/qvm_pool.py
View File

@ -66,6 +66,7 @@ class _Info(qubes.tools.PoolsAction):
def __init__(self, option_strings, help='print pool info and exit',
**kwargs):
# pylint: disable=redefined-builtin
super(_Info, self).__init__(option_strings, help=help, **kwargs)
def __call__(self, parser, namespace, values, option_string=None):

2
qubes/utils.py
View File

@ -26,13 +26,13 @@
import hashlib
import os
import pkg_resources
import re
import subprocess
import docutils
import docutils.core
import docutils.io
import pkg_resources
import qubes.exc

2
qubes/vm/dispvm.py
View File

@ -1,8 +1,6 @@
#!/usr/bin/python2 -O
# vim: fileencoding=utf-8
import random
import qubes.vm.qubesvm
import qubes.vm.appvm
import qubes.config

16
qubes/vm/mix/net.py
View File

@ -24,15 +24,17 @@
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import libvirt
import lxml.etree
import re
import lxml.etree
import libvirt
import qubes
import qubes.events
import qubes.exc
def _setter_mac(self, prop, value):
# pylint: disable=unused-argument
if not isinstance(value, basestring):
raise ValueError('MAC address must be a string')
value = value.lower()
@ -154,7 +156,7 @@ class NetVMMixin(qubes.events.Emitter):
in its netvm.
This is needed when starting netvm *after* its connected domains.
'''
''' # pylint: disable=unused-argument
if self.netvm:
self.netvm.reload_firewall_for_vm(self)
@ -170,7 +172,7 @@ class NetVMMixin(qubes.events.Emitter):
# 1426
vm.run('modprobe -r xen-netfront xennet',
user='root', wait=True)
except:
except: # pylint: disable=bare-except
pass
try:
@ -181,9 +183,11 @@ class NetVMMixin(qubes.events.Emitter):
@qubes.events.handler('domain-pre-shutdown')
def shutdown_net(self, event, force=False):
# pylint: disable=unused-argument
connected_vms = [vm for vm in self.connected_vms if vm.is_running()]
if connected_vms and not force:
raise qubes.exc.QubesVMError(
raise qubes.exc.QubesVMError(self,
'There are other VMs connected to this VM: {}'.format(
', '.join(vm.name for vm in connected_vms)))
@ -283,6 +287,7 @@ class NetVMMixin(qubes.events.Emitter):
@qubes.events.handler('property-pre-set:netvm')
def on_property_pre_set_netvm(self, event, name, new_netvm, old_netvm=None):
# pylint: disable=unused-argument
if new_netvm is None:
return
@ -334,5 +339,6 @@ class NetVMMixin(qubes.events.Emitter):
# FIXME use event after creating Xen domain object, but before "resume"
@qubes.events.handler('firewall-changed')
def on_firewall_changed(self, event):
# pylint: disable=unused-argument
if self.is_running() and self.netvm:
self.netvm.reload_firewall_for_vm(self)

82
qubes/vm/qubesvm.py
View File

@ -29,7 +29,6 @@ from __future__ import absolute_import
import base64
import datetime
import itertools
import lxml
import os
import os.path
import re
@ -40,26 +39,30 @@ import time
import uuid
import warnings
import lxml
import libvirt
import qubes
import qubes.config
import qubes.exc
import qubes.storage
import qubes.storage.domain
import qubes.storage.file
import qubes.tools.qvm_ls
import qubes.utils
import qubes.vm
import qubes.vm.mix.net
import qubes.tools.qvm_ls
from qubes.storage.domain import DomainPool
qmemman_present = False
try:
import qubes.qmemman.client
import qubes.qmemman.client # pylint: disable=wrong-import-position
qmemman_present = True
except ImportError:
pass
MEM_OVERHEAD_BASE = (3 + 1) * 1024 * 1024
MEM_OVERHEAD_PER_VCPU = 3 * 1024 * 1024 / 2
def _setter_qid(self, prop, value):
# pylint: disable=unused-argument
@ -132,6 +135,7 @@ def _setter_label(self, prop, value):
return self.app.get_label(value)
def _setter_positive_int(self, prop, value):
# pylint: disable=unused-argument
value = int(value)
if value <= 0:
raise ValueError('Value must be positive')
@ -225,7 +229,9 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
default='default',
doc='storage pool for this qube devices')
dir_path = property( (lambda self: os.path.join(qubes.config.system_path['qubes_base_dir'], self.dir_path_prefix, self.name)),
dir_path = property('dir_path',
(lambda self: os.path.join(qubes.config.system_path['qubes_base_dir'],
self.dir_path_prefix, self.name)),
doc='Root directory for files related to this domain')
# XXX swallowed uses_default_kernel
@ -359,7 +365,7 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
'''QubesDB handle for this domain.'''
if self._qdb_connection is None:
if self.is_running():
import qubesdb
import qubesdb # pylint: disable=import-error
self._qdb_connection = qubesdb.QubesDB(self.name)
return self._qdb_connection
@ -413,13 +419,13 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
def is_template(self):
warnings.warn('vm.is_template() is deprecated, use isinstance()',
DeprecationWarning)
import qubes.vm.templatevm
import qubes.vm.templatevm # pylint: disable=redefined-outer-name
return isinstance(self, qubes.vm.templatevm.TemplateVM)
def is_appvm(self):
warnings.warn('vm.is_appvm() is deprecated, use isinstance()',
DeprecationWarning)
import qubes.vm.appvm
import qubes.vm.appvm # pylint: disable=redefined-outer-name
return isinstance(self, qubes.vm.appvm.AppVM)
def is_proxyvm(self):
@ -430,7 +436,7 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
def is_disposablevm(self):
warnings.warn('vm.is_disposable() is deprecated, use isinstance()',
DeprecationWarning)
import qubes.vm.dispvm
import qubes.vm.dispvm # pylint: disable=redefined-outer-name
return isinstance(self, qubes.vm.dispvm.DispVM)
def is_netvm(self):
@ -447,20 +453,25 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
# constructor
#
def __init__(self, app, xml, volume_config={}, **kwargs):
def __init__(self, app, xml, volume_config=None, **kwargs):
super(QubesVM, self).__init__(app, xml, **kwargs)
self.volumes = {}
self.storage = None
self.volume_config = {}
if volume_config is None:
volume_config = {}
if hasattr(self, 'volume_config'):
if xml is not None:
for node in xml.xpath('volume-config/volume'):
name = node.get('name')
assert name
for k, v in node.items():
self.volume_config[name][k] = v
for key, value in node.items():
self.volume_config[name][key] = value
for name, conf in volume_config.items():
for k, v in conf.items():
self.volume_config[name][k] = v
for key, value in conf.items():
self.volume_config[name][key] = value
elif volume_config:
raise TypeError(
'volume_config specified, but {} did not expect that.'.format(
@ -472,7 +483,6 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
self._libvirt_domain = None
self._qdb_connection = None
if xml is None:
# we are creating new VM and attributes came through kwargs
assert hasattr(self, 'qid')
@ -510,14 +520,13 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
def __xml__(self):
element = super(QubesVM, self).__xml__()
if hasattr(self, 'volumes'):
volume_config_node = lxml.etree.Element('volume-config')
for volume in self.volumes.values():
volume_config_node.append(volume.__xml__())
element.append(volume_config_node)
return element
#
@ -531,8 +540,9 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
self.uuid = uuid.uuid4()
# Initialize VM image storage class
# XXX why not in constructor?
self.storage = qubes.storage.Storage(self)
vm_pool = DomainPool(self)
vm_pool = qubes.storage.domain.DomainPool(self)
self.app.pools[vm_pool.name] = vm_pool
@ -1019,9 +1029,6 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
if not qmemman_present:
return
MEM_OVERHEAD_BASE = (3 + 1) * 1024 * 1024
MEM_OVERHEAD_PER_VCPU = 3 * 1024 * 1024 / 2
if mem_required is None:
mem_required = int(self.memory) * 1024 * 1024
@ -1143,6 +1150,7 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
# FIXME move this to qubes.storage.xen.XenVMStorage
retcode = 0
if self.is_running():
# pylint: disable=redefined-variable-type
retcode = self.run('''
while [ "`blockdev --getsize64 /dev/xvdb`" -lt {0} ]; do
head /dev/xvdb >/dev/null;
@ -1408,7 +1416,7 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
'''Check whether domain is running and sane.
Currently this checks for running qrexec.
'''
''' # pylint: disable=unused-argument
# Running gui-daemon implies also VM running
if not self.is_qrexec_running():
@ -1531,10 +1539,10 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
.. seealso:: :py:meth:`get_root_img_sz`
'''
warnings.warn(
"get_disk_utilization_root_img is deprecated, use volumes['root'].utilization",
DeprecationWarning)
return qubes.storage.get_disk_usage(self.volumes['root'].utilization)
warnings.warn("get_disk_utilization_root_img is deprecated,"
" use volumes['root'].utilization", DeprecationWarning)
return qubes.storage.file.get_disk_usage(
self.volumes['root'].utilization)
# XXX shouldn't this go only to vms that have root image?
@ -1552,7 +1560,7 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
warnings.warn(
"get_disk_root_img_sz is deprecated, use volumes['root'].size",
DeprecationWarning)
return qubes.storage.get_disk_usage(self.volumes['root'].size)
return qubes.storage.file.get_disk_usage(self.volumes['root'].size)
def get_disk_utilization_private_img(self):
'''Get space that is actually ocuppied by :py:attr:`volumes['private']`.
@ -1561,12 +1569,11 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
call directly :py:attr:`volumes[name].utilization`
:returns: domain's real disk image size [FIXME unit]
:rtype: FIXME
'''
''' # pylint: disable=invalid-name
warnings.warn(
"get_disk_utilization_private_img is deprecated, use volumes['private'].utilization",
DeprecationWarning)
return qubes.storage.get_disk_usage(self.volumes[
warnings.warn("get_disk_utilization_private_img is deprecated,"
" use volumes['private'].utilization", DeprecationWarning)
return qubes.storage.file.get_disk_usage(self.volumes[
'private'].utilization)
def get_private_img_sz(self):
@ -1580,10 +1587,9 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
.. seealso:: :py:meth:`get_disk_utilization_private_img`
'''
warnings.warn(
"get_disk_private_img_sz is deprecated, use volumes['private'].size",
DeprecationWarning)
return qubes.storage.get_disk_usage(self.volumes['private'].size)
warnings.warn("get_disk_private_img_sz is deprecated,"
" use volumes['private'].size", DeprecationWarning)
return qubes.storage.file.get_disk_usage(self.volumes['private'].size)
def get_disk_utilization(self):
'''Return total space actually occuppied by all files belonging to \
@ -1593,7 +1599,7 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
:rtype: FIXME
'''
return qubes.storage.get_disk_usage(self.dir_path)
return qubes.storage.file.get_disk_usage(self.dir_path)
# TODO move to storage
def verify_files(self):