Add qvm-pool and qvm-volume tool
This commit is contained in:
Marek Marczykowski-Górecki
parent
1fd09f4e58
commit
36d8ee9b32
94
qubesmgmt/tests/tools/qvm_pool.py
Normal file
94
qubesmgmt/tests/tools/qvm_pool.py
Normal file
@ -0,0 +1,94 @@
|
||||
# -*- encoding: utf8 -*-
|
||||
#
|
||||
# The Qubes OS Project, http://www.qubes-os.org
|
||||
#
|
||||
# Copyright (C) 2017 Marek Marczykowski-Górecki
|
||||
# <marmarek@invisiblethingslab.com>
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation; either version 2.1 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Lesser General Public License along
|
||||
# with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import qubesmgmt.tests
|
||||
import qubesmgmt.tests.tools
|
||||
import qubesmgmt.tools.qvm_pool
|
||||
|
||||
|
||||
class TC_00_qvm_pool(qubesmgmt.tests.QubesTestCase):
|
||||
def test_000_list(self):
|
||||
self.app.expected_calls[('dom0', 'mgmt.pool.List', None, None)] = \
|
||||
b'0\x00pool-file\npool-lvm\n'
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.pool.Info', 'pool-file', None)] = \
|
||||
b'0\x00driver=file\ndir_path=/var/lib/qubes\n'
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.pool.Info', 'pool-lvm', None)] = \
|
||||
b'0\x00driver=lvm\nvolume_group=qubes_dom0\nthin_pool=pool00\n'
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_pool.main(['-l'], app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'NAME DRIVER\n'
|
||||
'pool-file file\n'
|
||||
'pool-lvm lvm\n')
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_010_list_drivers(self):
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.pool.ListDrivers', None, None)] = \
|
||||
b'0\x00file dir_path revisions_to_keep\n' \
|
||||
b'lvm volume_group thin_pool revisions_to_keep\n'
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_pool.main(['--help-drivers'], app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'DRIVER OPTIONS\n'
|
||||
'file dir_path, revisions_to_keep\n'
|
||||
'lvm volume_group, thin_pool, revisions_to_keep\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_020_add(self):
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.pool.Add', 'file',
|
||||
b'name=test-pool\ndir_path=/some/path\n')] = b'0\x00'
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_pool.main(
|
||||
['--add', 'test-pool', 'file', '-o', 'dir_path=/some/path'],
|
||||
app=self.app))
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_030_remove(self):
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.pool.Remove', 'test-pool', None)] = b'0\x00'
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_pool.main(['--remove', 'test-pool'],
|
||||
app=self.app))
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_040_info(self):
|
||||
self.app.expected_calls[('dom0', 'mgmt.pool.List', None, None)] = \
|
||||
b'0\x00pool-file\npool-lvm\n'
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.pool.Info', 'pool-lvm', None)] = \
|
||||
b'0\x00driver=lvm\nvolume_group=qubes_dom0\nthin_pool=pool00\n'
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_pool.main(['-i', 'pool-lvm'],
|
||||
app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'name pool-lvm\n'
|
||||
'driver lvm\n'
|
||||
'thin_pool pool00\n'
|
||||
'volume_group qubes_dom0\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
167
qubesmgmt/tests/tools/qvm_volume.py
Normal file
167
qubesmgmt/tests/tools/qvm_volume.py
Normal file
@ -0,0 +1,167 @@
|
||||
# -*- encoding: utf8 -*-
|
||||
#
|
||||
# The Qubes OS Project, http://www.qubes-os.org
|
||||
#
|
||||
# Copyright (C) 2017 Marek Marczykowski-Górecki
|
||||
# <marmarek@invisiblethingslab.com>
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation; either version 2.1 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Lesser General Public License along
|
||||
# with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import qubesmgmt.tests
|
||||
import qubesmgmt.tests.tools
|
||||
import qubesmgmt.tools.qvm_volume
|
||||
|
||||
|
||||
class TC_00_qvm_volume(qubesmgmt.tests.QubesTestCase):
|
||||
|
||||
def setup_expected_calls_for_list(self, vms=('vm1', 'sys-net')):
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.vm.List', None, None)] = \
|
||||
b'0\x00vm1 class=AppVM state=Running\n' \
|
||||
b'sys-net class=AppVM state=Running\n'
|
||||
for vm in vms:
|
||||
for vol in ('root', 'private'):
|
||||
self.app.expected_calls[
|
||||
(vm, 'mgmt.vm.volume.Info', vol, None)] = \
|
||||
b'0\x00' + \
|
||||
(b'pool=pool-file\n' if vol == 'root' else
|
||||
b'pool=other-pool\n') + \
|
||||
b'vid=' + vm.encode() + b'-' + vol.encode() + b'\n' \
|
||||
b'internal=True\n' \
|
||||
b'size=10737418240\n'
|
||||
self.app.expected_calls[
|
||||
(vm, 'mgmt.vm.volume.ListSnapshots', vol, None)] = \
|
||||
b'0\x00snap1\n' if vol == 'private' else b'0\x00'
|
||||
self.app.expected_calls[
|
||||
(vm, 'mgmt.vm.volume.List', None, None)] = \
|
||||
b'0\x00root\nprivate\n'
|
||||
|
||||
def test_000_list(self):
|
||||
self.setup_expected_calls_for_list()
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_volume.main(['ls', '-i'], app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'POOL:VOLUME VMNAME VOLUME_NAME '
|
||||
'REVERT_POSSIBLE\n'
|
||||
'other-pool:sys-net-private sys-net private Yes\n'
|
||||
'other-pool:vm1-private vm1 private Yes\n'
|
||||
'pool-file:sys-net-root sys-net root No\n'
|
||||
'pool-file:vm1-root vm1 root No\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_001_list_domain(self):
|
||||
self.setup_expected_calls_for_list(vms=('vm1',))
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_volume.main(['ls', '-i', 'vm1'],
|
||||
app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
|
||||
'other-pool:vm1-private vm1 private Yes\n'
|
||||
'pool-file:vm1-root vm1 root No\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_002_list_domain_pool(self):
|
||||
self.setup_expected_calls_for_list(vms=('vm1',))
|
||||
self.app.expected_calls[('dom0', 'mgmt.pool.List', None, None)] = \
|
||||
b'0\x00pool-file\nother-pool\n'
|
||||
del self.app.expected_calls[
|
||||
('vm1', 'mgmt.vm.volume.ListSnapshots', 'private', None)]
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_volume.main(
|
||||
['ls', '-i', '-p', 'pool-file', 'vm1'],
|
||||
app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
|
||||
'pool-file:vm1-root vm1 root No\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_003_list_pool(self):
|
||||
self.setup_expected_calls_for_list()
|
||||
self.app.expected_calls[('dom0', 'mgmt.pool.List', None, None)] = \
|
||||
b'0\x00pool-file\nother-pool\n'
|
||||
del self.app.expected_calls[
|
||||
('vm1', 'mgmt.vm.volume.ListSnapshots', 'private', None)]
|
||||
del self.app.expected_calls[
|
||||
('sys-net', 'mgmt.vm.volume.ListSnapshots', 'private', None)]
|
||||
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_volume.main(
|
||||
['ls', '-i', '-p', 'pool-file'],
|
||||
app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
|
||||
'pool-file:sys-net-root sys-net root No\n'
|
||||
'pool-file:vm1-root vm1 root No\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_004_list_multiple_domains(self):
|
||||
self.setup_expected_calls_for_list(vms=('vm1', 'vm2'))
|
||||
self.app.expected_calls[
|
||||
('dom0', 'mgmt.vm.List', None, None)] = \
|
||||
b'0\x00vm1 class=AppVM state=Running\n' \
|
||||
b'vm2 class=AppVM state=Running\n' \
|
||||
b'vm3 class=AppVM state=Running\n'
|
||||
with qubesmgmt.tests.tools.StdoutBuffer() as stdout:
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_volume.main(
|
||||
['ls', '-i', 'vm1', 'vm2'], app=self.app))
|
||||
self.assertEqual(stdout.getvalue(),
|
||||
'POOL:VOLUME VMNAME VOLUME_NAME REVERT_POSSIBLE\n'
|
||||
'other-pool:vm1-private vm1 private Yes\n'
|
||||
'other-pool:vm2-private vm2 private Yes\n'
|
||||
'pool-file:vm1-root vm1 root No\n'
|
||||
'pool-file:vm2-root vm2 root No\n'
|
||||
)
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_010_extend(self):
|
||||
self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
|
||||
b'0\x00testvm class=AppVM state=Running\n'
|
||||
self.app.expected_calls[
|
||||
('testvm', 'mgmt.vm.volume.List', None, None)] = \
|
||||
b'0\x00root\nprivate\n'
|
||||
self.app.expected_calls[
|
||||
('testvm', 'mgmt.vm.volume.Resize', 'private', b'10737418240')] = \
|
||||
b'0\x00'
|
||||
self.assertEqual(0,
|
||||
qubesmgmt.tools.qvm_volume.main(
|
||||
['extend', 'testvm:private', '10GiB'],
|
||||
app=self.app))
|
||||
self.assertAllCalled()
|
||||
|
||||
def test_011_extend_error(self):
|
||||
self.app.expected_calls[('dom0', 'mgmt.vm.List', None, None)] = \
|
||||
b'0\x00testvm class=AppVM state=Running\n'
|
||||
self.app.expected_calls[
|
||||
('testvm', 'mgmt.vm.volume.List', None, None)] = \
|
||||
b'0\x00root\nprivate\n'
|
||||
self.app.expected_calls[
|
||||
('testvm', 'mgmt.vm.volume.Resize', 'private', b'1073741824')] = \
|
||||
b'2\x00StoragePoolException\x00\x00Failed to resize volume: ' \
|
||||
b'shrink not allowed\x00'
|
||||
with qubesmgmt.tests.tools.StderrBuffer() as stderr:
|
||||
self.assertEqual(1,
|
||||
qubesmgmt.tools.qvm_volume.main(
|
||||
['extend', 'testvm:private', '1GiB'],
|
||||
app=self.app))
|
||||
self.assertIn('shrink not allowed', stderr.getvalue())
|
||||
self.assertAllCalled()
|
||||
@ -229,6 +229,45 @@ class VolumeAction(QubesAction):
|
||||
parser.error('expected a pool & volume id combination like foo:bar')
|
||||
|
||||
|
||||
class VMVolumeAction(QubesAction):
|
||||
''' Action for argument parser that gets the
|
||||
:py:class:``qubes.storage.Volume`` from a VM:VOLUME string.
|
||||
'''
|
||||
# pylint: disable=too-few-public-methods
|
||||
|
||||
def __init__(self, help='A pool & volume id combination',
|
||||
required=True, **kwargs):
|
||||
# pylint: disable=redefined-builtin
|
||||
super(VMVolumeAction, self).__init__(help=help, required=required,
|
||||
**kwargs)
|
||||
|
||||
def __call__(self, parser, namespace, values, option_string=None):
|
||||
''' Set ``namespace.vmname`` to ``values`` '''
|
||||
setattr(namespace, self.dest, values)
|
||||
|
||||
def parse_qubes_app(self, parser, namespace):
|
||||
''' Acquire the :py:class:``qubes.storage.Volume`` object from
|
||||
``namespace.app``.
|
||||
'''
|
||||
assert hasattr(namespace, 'app')
|
||||
app = namespace.app
|
||||
|
||||
try:
|
||||
vm_name, vol_name = getattr(namespace, self.dest).split(':')
|
||||
try:
|
||||
vm = app.domains[vm_name]
|
||||
try:
|
||||
volume = vm.volumes[vol_name]
|
||||
setattr(namespace, self.dest, volume)
|
||||
except KeyError:
|
||||
parser.error_runtime('vm {!r} has no volume {!r}'.format(
|
||||
vm_name, vol_name))
|
||||
except KeyError:
|
||||
parser.error_runtime('no vm {!r}'.format(vm_name))
|
||||
except ValueError:
|
||||
parser.error('expected a vm & volume combination like foo:bar')
|
||||
|
||||
|
||||
class PoolsAction(QubesAction):
|
||||
''' Action for argument parser to gather multiple pools '''
|
||||
# pylint: disable=too-few-public-methods
|
||||
@ -247,7 +286,7 @@ class PoolsAction(QubesAction):
|
||||
pool_names = getattr(namespace, self.dest)
|
||||
if pool_names:
|
||||
try:
|
||||
pools = [app.get_pool(name) for name in pool_names]
|
||||
pools = [app.pools[name] for name in pool_names]
|
||||
setattr(namespace, self.dest, pools)
|
||||
except qubesmgmt.exc.QubesException as e:
|
||||
parser.error(str(e))
|
||||
|
||||
178
qubesmgmt/tools/qvm_pool.py
Normal file
178
qubesmgmt/tools/qvm_pool.py
Normal file
@ -0,0 +1,178 @@
|
||||
# pylint: disable=too-few-public-methods
|
||||
|
||||
#
|
||||
# The Qubes OS Project, http://www.qubes-os.org
|
||||
#
|
||||
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation; either version 2.1 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Lesser General Public License along
|
||||
# with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
'''Manages Qubes pools and their options'''
|
||||
|
||||
from __future__ import print_function
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
import qubesmgmt
|
||||
import qubesmgmt.exc
|
||||
import qubesmgmt.storage
|
||||
import qubesmgmt.tools
|
||||
|
||||
|
||||
class _Info(qubesmgmt.tools.PoolsAction):
|
||||
''' Action for argument parser that displays pool info and exits. '''
|
||||
|
||||
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):
|
||||
setattr(namespace, 'command', 'info')
|
||||
super(_Info, self).__call__(parser, namespace, values, option_string)
|
||||
|
||||
|
||||
def pool_info(pool):
|
||||
''' Prints out pool name and config '''
|
||||
data = [("name", pool.name)]
|
||||
data += [i for i in sorted(pool.config.items()) if i[0] != 'name']
|
||||
qubesmgmt.tools.print_table(data)
|
||||
|
||||
|
||||
def list_pools(app):
|
||||
''' Prints out all known pools and their drivers '''
|
||||
result = [('NAME', 'DRIVER')]
|
||||
for pool in app.pools:
|
||||
result += [(pool.name, pool.driver)]
|
||||
qubesmgmt.tools.print_table(result)
|
||||
|
||||
|
||||
class _Remove(argparse.Action):
|
||||
''' Action for argument parser that removes a pool '''
|
||||
|
||||
def __init__(self, option_strings, dest=None, default=None, metavar=None):
|
||||
super(_Remove, self).__init__(option_strings=option_strings,
|
||||
dest=dest,
|
||||
metavar=metavar,
|
||||
default=default,
|
||||
help='remove pool')
|
||||
|
||||
def __call__(self, parser, namespace, name, option_string=None):
|
||||
setattr(namespace, 'command', 'remove')
|
||||
setattr(namespace, 'name', name)
|
||||
|
||||
|
||||
class _Add(argparse.Action):
|
||||
''' Action for argument parser that adds a pool. '''
|
||||
|
||||
def __init__(self, option_strings, dest=None, default=None, metavar=None):
|
||||
super(_Add, self).__init__(option_strings=option_strings,
|
||||
dest=dest,
|
||||
metavar=metavar,
|
||||
default=default,
|
||||
nargs=2,
|
||||
help='add pool')
|
||||
|
||||
def __call__(self, parser, namespace, values, option_string=None):
|
||||
name, driver = values
|
||||
setattr(namespace, 'command', 'add')
|
||||
setattr(namespace, 'name', name)
|
||||
setattr(namespace, 'driver', driver)
|
||||
|
||||
|
||||
class _Options(argparse.Action):
|
||||
''' Action for argument parser that parsers options. '''
|
||||
|
||||
def __init__(self, option_strings, dest, default, metavar='options'):
|
||||
super(_Options, self).__init__(
|
||||
option_strings=option_strings,
|
||||
dest=dest,
|
||||
metavar=metavar,
|
||||
default=default,
|
||||
help='comma-separated list of driver options')
|
||||
|
||||
def __call__(self, parser, namespace, options, option_string=None):
|
||||
setattr(namespace, 'options',
|
||||
dict([option.split('=', 1) for option in options.split(',')]))
|
||||
|
||||
|
||||
def get_parser():
|
||||
''' Parses the provided args '''
|
||||
parser = qubesmgmt.tools.QubesArgumentParser(description=__doc__)
|
||||
parser.add_argument('-o', action=_Options, dest='options', default={})
|
||||
group = parser.add_mutually_exclusive_group()
|
||||
group.add_argument('-l',
|
||||
'--list',
|
||||
dest='command',
|
||||
const='list',
|
||||
action='store_const',
|
||||
help='list all pools and exit (default action)')
|
||||
group.add_argument('-i', '--info', metavar='POOLNAME', dest='pools',
|
||||
action=_Info, default=[])
|
||||
group.add_argument('-a',
|
||||
'--add',
|
||||
action=_Add,
|
||||
dest='command',
|
||||
metavar=('NAME', 'DRIVER'))
|
||||
group.add_argument('-r', '--remove', metavar='NAME', action=_Remove)
|
||||
group.add_argument('--help-drivers',
|
||||
dest='command',
|
||||
const='list-drivers',
|
||||
action='store_const',
|
||||
help='list all drivers with their options and exit')
|
||||
return parser
|
||||
|
||||
|
||||
def main(args=None, app=None):
|
||||
'''Main routine of :program:`qvm-pools`.
|
||||
|
||||
:param list args: Optional arguments to override those delivered from \
|
||||
command line.
|
||||
'''
|
||||
parser = get_parser()
|
||||
try:
|
||||
args = parser.parse_args(args, app=app)
|
||||
except qubesmgmt.exc.QubesException as e:
|
||||
parser.print_error(str(e))
|
||||
return 1
|
||||
|
||||
if args.command is None or args.command == 'list':
|
||||
list_pools(args.app)
|
||||
elif args.command == 'list-drivers':
|
||||
result = [('DRIVER', 'OPTIONS')]
|
||||
for driver in sorted(args.app.pool_drivers):
|
||||
params = args.app.pool_driver_parameters(driver)
|
||||
driver_options = ', '.join(params)
|
||||
result += [(driver, driver_options)]
|
||||
qubesmgmt.tools.print_table(result)
|
||||
elif args.command == 'add':
|
||||
try:
|
||||
args.app.add_pool(name=args.name, driver=args.driver,
|
||||
**args.options)
|
||||
except qubesmgmt.exc.QubesException as e:
|
||||
parser.error('failed to add pool %s: %s\n' % (args.name, str(e)))
|
||||
elif args.command == 'remove':
|
||||
try:
|
||||
args.app.remove_pool(args.name)
|
||||
except KeyError:
|
||||
parser.print_error('no such pool %s\n' % args.name)
|
||||
elif args.command == 'info':
|
||||
for pool in args.pools:
|
||||
pool_info(pool)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
227
qubesmgmt/tools/qvm_volume.py
Normal file
227
qubesmgmt/tools/qvm_volume.py
Normal file
@ -0,0 +1,227 @@
|
||||
# encoding=utf-8
|
||||
#
|
||||
# The Qubes OS Project, http://www.qubes-os.org
|
||||
#
|
||||
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
|
||||
# Copyright (C) 2017 Marek Marczykowski-Górecki
|
||||
# <marmarek@invisiblethingslab.com>
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published by
|
||||
# the Free Software Foundation; either version 2.1 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Lesser General Public License along
|
||||
# with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
'''Qubes volume management'''
|
||||
|
||||
from __future__ import print_function
|
||||
|
||||
import sys
|
||||
|
||||
import qubesmgmt
|
||||
import qubesmgmt.exc
|
||||
import qubesmgmt.tools
|
||||
import qubesmgmt.utils
|
||||
|
||||
|
||||
def prepare_table(vd_list, full=False):
|
||||
''' Converts a list of :py:class:`VolumeData` objects to a list of tupples
|
||||
for the :py:func:`qubes.tools.print_table`.
|
||||
|
||||
If :program:`qvm-volume` is running in a TTY, it will ommit duplicate
|
||||
data.
|
||||
|
||||
:param list vd_list: List of :py:class:`VolumeData` objects.
|
||||
:param bool full: If set to true duplicate data is printed even when
|
||||
running from TTY.
|
||||
:returns: list of tupples
|
||||
'''
|
||||
output = []
|
||||
output += [('POOL:VOLUME', 'VMNAME', 'VOLUME_NAME', 'REVERT_POSSIBLE')]
|
||||
|
||||
for volume in sorted(vd_list):
|
||||
if volume.domains:
|
||||
vmname, volume_name = volume.domains.pop()
|
||||
output += [(str(volume), vmname, volume_name, volume.revisions)]
|
||||
for tupple in volume.domains:
|
||||
vmname, volume_name = tupple
|
||||
if full or not sys.stdout.isatty():
|
||||
output += [(str(volume), vmname, volume_name,
|
||||
volume.revisions)]
|
||||
else:
|
||||
output += [('', vmname, volume_name, volume.revisions)]
|
||||
else:
|
||||
output += [(str(volume), "")]
|
||||
|
||||
return output
|
||||
|
||||
|
||||
class VolumeData(object):
|
||||
''' Wrapper object around :py:class:`qubes.storage.Volume`, mainly to track
|
||||
the domains a volume is attached to.
|
||||
'''
|
||||
# pylint: disable=too-few-public-methods
|
||||
def __init__(self, volume):
|
||||
self.pool = volume.pool
|
||||
self.vid = volume.vid
|
||||
if len(volume.revisions) > 0:
|
||||
self.revisions = 'Yes'
|
||||
else:
|
||||
self.revisions = 'No'
|
||||
self.domains = []
|
||||
|
||||
def __lt__(self, other):
|
||||
return (self.pool, self.vid) < (other.pool, other.vid)
|
||||
|
||||
def __str__(self):
|
||||
return "{!s}:{!s}".format(self.pool, self.vid)
|
||||
|
||||
|
||||
def list_volumes(args):
|
||||
''' Called by the parser to execute the qvm-volume list subcommand. '''
|
||||
app = args.app
|
||||
|
||||
if hasattr(args, 'domains') and args.domains:
|
||||
domains = args.domains
|
||||
else:
|
||||
domains = app.domains
|
||||
volumes = [v for vm in domains for v in vm.volumes.values()]
|
||||
|
||||
if args.pools:
|
||||
# only specified pools
|
||||
volumes = [v for v in volumes if v.pool in args.pools]
|
||||
|
||||
if not args.internal: # hide internal volumes
|
||||
volumes = [v for v in volumes if not v.internal]
|
||||
|
||||
vd_dict = {}
|
||||
for volume in volumes:
|
||||
volume_data = VolumeData(volume)
|
||||
try:
|
||||
vd_dict[volume.pool][volume.vid] = volume_data
|
||||
except KeyError:
|
||||
vd_dict[volume.pool] = {volume.vid: volume_data}
|
||||
|
||||
for domain in domains: # gather the domain names
|
||||
try:
|
||||
for name, volume in domain.volumes.items():
|
||||
try:
|
||||
volume_data = vd_dict[volume.pool][volume.vid]
|
||||
volume_data.domains += [(domain.name, name)]
|
||||
except KeyError:
|
||||
# Skipping volume
|
||||
continue
|
||||
except AttributeError:
|
||||
# Skipping domain without volumes
|
||||
continue
|
||||
|
||||
if hasattr(args, 'domains') and args.domains:
|
||||
result = [x # reduce to only VolumeData with assigned domains
|
||||
for p in vd_dict.values() for x in p.values()
|
||||
if x.domains]
|
||||
else:
|
||||
result = [x for p in vd_dict.values() for x in p.values()]
|
||||
qubesmgmt.tools.print_table(prepare_table(result, full=args.full))
|
||||
|
||||
|
||||
def revert_volume(args):
|
||||
''' Revert volume to previous state '''
|
||||
volume = args.volume
|
||||
app = args.app
|
||||
try:
|
||||
pool = app.pools[volume.pool]
|
||||
pool.revert(volume)
|
||||
except qubesmgmt.exc.StoragePoolException as e:
|
||||
print(str(e), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def extend_volumes(args):
|
||||
''' Called by the parser to execute the :program:`qvm-block extend`
|
||||
subcommand
|
||||
'''
|
||||
volume = args.volume
|
||||
size = qubesmgmt.utils.parse_size(args.size)
|
||||
volume.resize(size)
|
||||
|
||||
|
||||
def init_list_parser(sub_parsers):
|
||||
''' Configures the parser for the :program:`qvm-block list` subcommand '''
|
||||
# pylint: disable=protected-access
|
||||
list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'),
|
||||
help='list storage volumes')
|
||||
list_parser.add_argument('-p', '--pool', dest='pools',
|
||||
action=qubesmgmt.tools.PoolsAction)
|
||||
list_parser.add_argument('-i', '--internal', action='store_true',
|
||||
help='Show internal volumes')
|
||||
list_parser.add_argument(
|
||||
'--full', action='store_true',
|
||||
help='print full line for each POOL_NAME:VOLUME_ID & vm combination')
|
||||
|
||||
vm_name_group = qubesmgmt.tools.VmNameGroup(
|
||||
list_parser, required=False, vm_action=qubesmgmt.tools.VmNameAction,
|
||||
help='list volumes from specified domain(s)')
|
||||
list_parser._mutually_exclusive_groups.append(vm_name_group)
|
||||
list_parser.set_defaults(func=list_volumes)
|
||||
|
||||
|
||||
def init_revert_parser(sub_parsers):
|
||||
''' Add 'revert' action related options '''
|
||||
revert_parser = sub_parsers.add_parser(
|
||||
'revert', aliases=('rv', 'r'),
|
||||
help='revert volume to previous revision')
|
||||
revert_parser.add_argument(metavar='VM:VOLUME', dest='volume',
|
||||
action=qubesmgmt.tools.VMVolumeAction)
|
||||
revert_parser.set_defaults(func=revert_volume)
|
||||
|
||||
|
||||
def init_extend_parser(sub_parsers):
|
||||
''' Add 'extend' action related options '''
|
||||
extend_parser = sub_parsers.add_parser(
|
||||
"extend", help="extend volume from domain", aliases=('d', 'dt'))
|
||||
extend_parser.add_argument(metavar='VM:VOLUME', dest='volume',
|
||||
action=qubesmgmt.tools.VMVolumeAction)
|
||||
extend_parser.add_argument('size', help='New size in bytes')
|
||||
extend_parser.set_defaults(func=extend_volumes)
|
||||
|
||||
def get_parser():
|
||||
'''Create :py:class:`argparse.ArgumentParser` suitable for
|
||||
:program:`qvm-block`.
|
||||
'''
|
||||
parser = qubesmgmt.tools.QubesArgumentParser(description=__doc__,
|
||||
want_app=True)
|
||||
parser.register('action', 'parsers',
|
||||
qubesmgmt.tools.AliasedSubParsersAction)
|
||||
sub_parsers = parser.add_subparsers(
|
||||
title='commands',
|
||||
description="For more information see qvm-block command -h",
|
||||
dest='command')
|
||||
init_extend_parser(sub_parsers)
|
||||
init_list_parser(sub_parsers)
|
||||
init_revert_parser(sub_parsers)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def main(args=None, app=None):
|
||||
'''Main routine of :program:`qvm-block`.'''
|
||||
parser = get_parser()
|
||||
try:
|
||||
args = parser.parse_args(args, app=app)
|
||||
args.func(args)
|
||||
except qubesmgmt.exc.QubesException as e:
|
||||
parser.print_error(str(e))
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Loading…
Reference in New Issue
Block a user