qvm-volume: add 'info' and 'config' actions

This allows to get and set volumes properties.

Fixes QubesOS/qubes-issues#3256

doc/manpages/qvm-volume.rst

@@ -60,6 +60,30 @@ passed or stdout is redirected to a pipe or file.
 
 aliases: ls, l
 
+info
+^^^^
+| :command:`qvm-volume info` [-h] [--verbose] [--quiet] *VMNAME:VOLUME* [*PROPERTY*]
+
+Show information about given volume - all properties and available revisions
+(for `revert` action). If specific property is given, only its value is printed.
+For list of revisions use `revisions` value.
+
+aliases: i
+
+config
+^^^^^^
+| :command:`qvm-volume config` [-h] [--verbose] [--quiet] *VMNAME:VOLUME* *PROPERTY* *VALUE*
+
+Set property of given volume. Properties currently possible to change:
+
+  - `rw` - `True` if volume should be writeable by the qube, `False` otherwise
+  - `revisions_to_keep` - how many revisions (previous versions of volume)
+   should be keep. At each qube shutdown its previous state is saved in new
+   revision, and the oldest revisions are remove so that only
+   `revisions_to_keep` are left. Set to `0` to not leave any previous versions.
+
+aliases: c, set, s
+
 extend
 ^^^^^^
 | :command:`qvm-volume extend` [-h] [--verbose] [--quiet] *POOL_NAME:VOLUME_ID* *NEW_SIZE*

qubesadmin/tests/tools/qvm_volume.py

@@ -250,3 +250,175 @@ class TC_00_qvm_volume(qubesadmin.tests.QubesTestCase):
                 ['revert', 'testvm:private', '20050101'],
                 app=self.app))
         self.assertAllCalled()
+
+    def test_030_set_revisions_to_keep(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.Set.revisions_to_keep', 'private',
+            b'3')] = b'0\x00'
+        self.assertEqual(0,
+            qubesadmin.tools.qvm_volume.main(
+                ['set', 'testvm:private', 'revisions_to_keep', '3'],
+                app=self.app))
+        self.assertAllCalled()
+
+    def test_031_set_rw(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.Set.rw', 'private',
+            b'True')] = b'0\x00'
+        self.assertEqual(0,
+            qubesadmin.tools.qvm_volume.main(
+                ['set', 'testvm:private', 'rw', 'True'],
+                app=self.app))
+        self.assertAllCalled()
+
+    def test_032_set_invalid(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.assertNotEqual(0,
+            qubesadmin.tools.qvm_volume.main(
+                ['set', 'testvm:private', 'invalid', 'True'],
+                app=self.app))
+        self.assertAllCalled()
+
+    def test_040_info(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.Info', 'private', None)] = \
+            b'0\x00pool=lvm\n' \
+            b'vid=qubes_dom0/vm-testvm-private\n' \
+            b'size=2147483648\n' \
+            b'usage=10000000\n' \
+            b'rw=True\n' \
+            b'source=\n' \
+            b'save_on_stop=True\n' \
+            b'snap_on_start=False\n' \
+            b'revisions_to_keep=3\n' \
+            b'is_outdated=False\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.ListSnapshots', 'private', None)] = \
+            b'0\x00200101010000\n200201010000\n200301010000\n'
+        with qubesadmin.tests.tools.StdoutBuffer() as stdout:
+            self.assertEqual(0,
+                qubesadmin.tools.qvm_volume.main(['info', 'testvm:private'],
+                    app=self.app))
+        output = stdout.getvalue()
+        # travis...
+        output = output.replace('\nsource\n', '\nsource             \n')
+        self.assertEqual(output,
+            'pool               lvm\n'
+            'vid                qubes_dom0/vm-testvm-private\n'
+            'rw                 True\n'
+            'source             \n'
+            'save_on_stop       True\n'
+            'snap_on_start      False\n'
+            'size               2147483648\n'
+            'usage              10000000\n'
+            'revisions_to_keep  3\n'
+            'is_outdated        False\n'
+            'Available revisions (for revert):\n'
+            '  200101010000\n'
+            '  200201010000\n'
+            '  200301010000\n')
+        self.assertAllCalled()
+
+    def test_041_info_no_revisions(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.Info', 'root', None)] = \
+            b'0\x00pool=lvm\n' \
+            b'vid=qubes_dom0/vm-testvm-root\n' \
+            b'size=2147483648\n' \
+            b'usage=10000000\n' \
+            b'rw=True\n' \
+            b'source=qubes_dom0/vm-fedora-26-root\n' \
+            b'save_on_stop=False\n' \
+            b'snap_on_start=True\n' \
+            b'revisions_to_keep=0\n' \
+            b'is_outdated=False\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.ListSnapshots', 'root', None)] = \
+            b'0\x00'
+        with qubesadmin.tests.tools.StdoutBuffer() as stdout:
+            self.assertEqual(0,
+                qubesadmin.tools.qvm_volume.main(['info', 'testvm:root'],
+                    app=self.app))
+        self.assertEqual(stdout.getvalue(),
+            'pool               lvm\n'
+            'vid                qubes_dom0/vm-testvm-root\n'
+            'rw                 True\n'
+            'source             qubes_dom0/vm-fedora-26-root\n'
+            'save_on_stop       False\n'
+            'snap_on_start      True\n'
+            'size               2147483648\n'
+            'usage              10000000\n'
+            'revisions_to_keep  0\n'
+            'is_outdated        False\n'
+            'Available revisions (for revert): none\n')
+        self.assertAllCalled()
+
+    def test_042_info_single_prop(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.Info', 'root', None)] = \
+            b'0\x00pool=lvm\n' \
+            b'vid=qubes_dom0/vm-testvm-root\n' \
+            b'size=2147483648\n' \
+            b'usage=10000000\n' \
+            b'rw=True\n' \
+            b'source=qubes_dom0/vm-fedora-26-root\n' \
+            b'save_on_stop=False\n' \
+            b'snap_on_start=True\n' \
+            b'revisions_to_keep=0\n' \
+            b'is_outdated=False\n'
+        with qubesadmin.tests.tools.StdoutBuffer() as stdout:
+            self.assertEqual(0,
+                qubesadmin.tools.qvm_volume.main(
+                    ['info', 'testvm:root', 'usage'],
+                    app=self.app))
+        self.assertEqual(stdout.getvalue(), '10000000\n')
+        self.assertAllCalled()
+
+    def test_043_info_revisions_only(self):
+        self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00testvm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\n'
+        self.app.expected_calls[
+            ('testvm', 'admin.vm.volume.ListSnapshots', 'private', None)] = \
+            b'0\x00200101010000\n200201010000\n200301010000\n'
+        with qubesadmin.tests.tools.StdoutBuffer() as stdout:
+            self.assertEqual(0,
+                qubesadmin.tools.qvm_volume.main(
+                    ['info', 'testvm:private', 'revisions'],
+                    app=self.app))
+        self.assertEqual(stdout.getvalue(),
+            '200101010000\n'
+            '200201010000\n'
+            '200301010000\n')
+        self.assertAllCalled()

qubesadmin/tools/qvm_volume.py

@@ -24,8 +24,11 @@
 
 from __future__ import print_function
 
+import argparse
 import sys
 
+import collections
+
 import qubesadmin
 import qubesadmin.exc
 import qubesadmin.tools
@@ -84,6 +87,50 @@ class VolumeData(object):
     def __str__(self):
         return "{!s}:{!s}".format(self.pool, self.vid)
 
+def info_volume(args):
+    ''' Show info about selected volume '''
+    volume = args.volume
+    info_items = ('pool', 'vid', 'rw', 'source', 'save_on_stop',
+        'snap_on_start', 'size', 'usage', 'revisions_to_keep')
+    if args.property:
+        if args.property == 'revisions':
+            for rev in volume.revisions:
+                print(rev)
+        elif args.property == 'is_outdated':
+            print(volume.is_outdated())
+        elif args.property in info_items:
+            value = getattr(volume, args.property)
+            if value is None:
+                value = ''
+            print(value)
+        else:
+            raise qubesadmin.exc.StoragePoolException(
+                'No such property: {}'.format(args.property))
+    else:
+        info = collections.OrderedDict()
+        for item in info_items:
+            value = getattr(volume, item)
+            if value is None:
+                value = ''
+            info[item] = str(value)
+        info['is_outdated'] = str(volume.is_outdated())
+
+        qubesadmin.tools.print_table(info.items())
+        revisions = volume.revisions
+        if revisions:
+            print('Available revisions (for revert):')
+            for rev in revisions:
+                print('  ' + rev)
+        else:
+            print('Available revisions (for revert): none')
+
+def config_volume(args):
+    ''' Change property of selected volume '''
+    volume = args.volume
+    if not args.property in ('rw', 'revisions_to_keep'):
+        raise qubesadmin.exc.QubesNoSuchPropertyError(
+            'Invalid property: {}'.format(args.property))
+    setattr(volume, args.property, args.value)
 
 def list_volumes(args):
     ''' Called by the parser to execute the qvm-volume list subcommand. '''
@@ -196,10 +243,32 @@ def init_extend_parser(sub_parsers):
     extend_parser.add_argument('size', help='New size in bytes')
     extend_parser.set_defaults(func=extend_volumes)
 
+def init_info_parser(sub_parsers):
+    ''' Add 'info' action related options '''
+    info_parser = sub_parsers.add_parser(
+        'info', aliases=('i',), help='info about volume')
+    info_parser.add_argument(metavar='VM:VOLUME', dest='volume',
+                             action=qubesadmin.tools.VMVolumeAction)
+    info_parser.add_argument(dest='property', action='store',
+        nargs=argparse.OPTIONAL,
+        help='Show only this property instead of all of them; use '
+             '\'revisions\' to list available revisions')
+    info_parser.set_defaults(func=info_volume)
+
+def init_config_parser(sub_parsers):
+    ''' Add 'info' action related options '''
+    info_parser = sub_parsers.add_parser(
+        'config', aliases=('c', 'set', 's'),
+        help='set config option for a volume')
+    info_parser.add_argument(metavar='VM:VOLUME', dest='volume',
+                             action=qubesadmin.tools.VMVolumeAction)
+    info_parser.add_argument(dest='property', action='store')
+    info_parser.add_argument(dest='value', action='store')
+    info_parser.set_defaults(func=config_volume)
 
 def get_parser():
     '''Create :py:class:`argparse.ArgumentParser` suitable for
-    :program:`qvm-block`.
+    :program:`qvm-volume`.
     '''
     parser = qubesadmin.tools.QubesArgumentParser(description=__doc__,
         want_app=True)
@@ -207,8 +276,10 @@ def get_parser():
         qubesadmin.tools.AliasedSubParsersAction)
     sub_parsers = parser.add_subparsers(
         title='commands',
-        description="For more information see qvm-block command -h",
+        description="For more information see qvm-volume command -h",
         dest='command')
+    init_info_parser(sub_parsers)
+    init_config_parser(sub_parsers)
     init_extend_parser(sub_parsers)
     init_list_parser(sub_parsers)
     init_revert_parser(sub_parsers)
@@ -219,7 +290,7 @@ def get_parser():
 
 
 def main(args=None, app=None):
-    '''Main routine of :program:`qvm-block`.'''
+    '''Main routine of :program:`qvm-volume`.'''
     parser = get_parser()
     try:
         args = parser.parse_args(args, app=app)