mgmt.vm.volume.* API support

QubesOS/qubes-issues#853

ci/pylintrc

@@ -54,7 +54,7 @@ variable-rgx=[a-z_][a-z0-9_]{2,30}$
 inlinevar-rgx=[A-Za-z_][A-Za-z0-9_]*$
 
 # Good variable names which should always be accepted, separated by a comma
-good-names=e,i,j,k,m,p,ex,Run,_,log,vm,ip,fd,fh
+good-names=e,i,j,k,m,p,rw,ex,Run,_,log,vm,ip,fd,fh
 
 # Bad variable names which should always be refused, separated by a comma
 bad-names=foo,bar,baz,toto,tutu,tata

qubesmgmt/storage.py

@@ -0,0 +1,174 @@
+# -*- 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/>.
+
+'''Storage subsystem.'''
+
+class Volume(object):
+    '''Storage volume.'''
+    def __init__(self, app, pool=None, vid=None, vm=None, vm_name=None):
+        '''Construct a Volume object.
+
+        Volume may be identified using pool+vid, or vm+vm_name. Either of
+        those argument pairs must be given.
+
+        :param Qubes app: application instance
+        :param str pool: pool name
+        :param str vid: volume id (within pool)
+        :param str vm: owner VM name
+        :param str vm_name: name within owning VM (like 'private', 'root' etc)
+        '''
+        self.app = app
+        if pool is None and vm is None:
+            raise ValueError('Either pool or vm must be given')
+        if pool is not None and vid is None:
+            raise ValueError('If pool is given, vid must be too.')
+        if vm is not None and vm_name is None:
+            raise ValueError('If vm is given, vm_name must be too.')
+        self._pool = pool
+        self._vid = vid
+        self._vm = vm
+        self._vm_name = vm_name
+        self._info = None
+
+    def _qubesd_call(self, func_name, payload=None):
+        '''Make a call to qubesd regarding this volume
+
+        :param str func_name: API function name, like `Info` or `Resize`
+        :param bytes payload: Payload to send.
+        '''
+        if self._vm is not None:
+            method = 'mgmt.vm.volume.' + func_name
+            dest = self._vm
+            arg = self._vm_name
+        else:
+            method = 'mgmt.pool.volume.' + func_name
+            dest = 'dom0'
+            # TODO: encode ':' and vid somehow
+            arg = self._pool + ':' + self._vid
+        return self.app.qubesd_call(dest, method, arg, payload)
+
+    def _fetch_info(self, force=True):
+        '''Fetch volume properties
+
+        Populate self._info dict
+
+        :param bool force: refresh self._info, even if already populated.
+        '''
+        if not force and self._info is not None:
+            return
+        info = self._qubesd_call('Info')
+        info = info.decode('ascii')
+        self._info = dict([line.split('=', 1) for line in info.splitlines()])
+
+    @property
+    def pool(self):
+        '''Storage volume pool name.'''
+        if self._pool is not None:
+            return self._pool
+        else:
+            self._fetch_info()
+            return self._info['pool']
+
+    @property
+    def vid(self):
+        '''Storage volume id, unique within given pool.'''
+        if self._vid is not None:
+            return self._vid
+        else:
+            self._fetch_info()
+            return self._info['vid']
+
+    @property
+    def size(self):
+        '''Size of volume, in bytes.'''
+        self._fetch_info(True)
+        return int(self._info['size'])
+
+    @property
+    def usage(self):
+        '''Used volume space, in bytes.'''
+        self._fetch_info(True)
+        return int(self._info['usage'])
+
+    @property
+    def rw(self):
+        '''True if volume is read-write.'''
+        self._fetch_info()
+        return self._info['rw'] == 'True'
+
+    @property
+    def snap_on_start(self):
+        '''Create a snapshot from source on VM start.'''
+        self._fetch_info()
+        return self._info['snap_on_start'] == 'True'
+
+    @property
+    def save_on_stop(self):
+        '''Commit changes to original volume on VM stop.'''
+        self._fetch_info()
+        return self._info['save_on_stop'] == 'True'
+
+    @property
+    def source(self):
+        '''Volume ID of source volume (for :py:attr:`snap_on_start`).
+
+        If None, this volume itself will be used.
+        '''
+        self._fetch_info()
+        if self._info['source']:
+            return self._info['source']
+        else:
+            return None
+
+    @property
+    def internal(self):
+        '''If `True` volume is hidden when qvm-block is used'''
+        self._fetch_info()
+        return self._info['internal'] == 'True'
+
+    @property
+    def revisions_to_keep(self):
+        '''Number of revisions to keep around'''
+        self._fetch_info()
+        return int(self._info['revisions_to_keep'])
+
+    def resize(self, size):
+        '''Resize volume.
+
+        Currently only extending is supported.
+
+        :param int size: new size in bytes.
+        '''
+        self._qubesd_call('Resize', str(size).encode('ascii'))
+
+    @property
+    def revisions(self):
+        ''' Returns iterable containing revision identifiers'''
+        revisions = self._qubesd_call('ListSnapshots')
+        return revisions.decode('ascii').splitlines()
+
+    def revert(self, revision):
+        ''' Revert volume to previous revision
+
+        :param str revision: Revision identifier to revert to
+        '''
+        if not isinstance(revision, str):
+            raise TypeError('revision must be a str')
+        self._qubesd_call('Revert', revision.encode('ascii'))

qubesmgmt/tests/storage.py

@@ -0,0 +1,234 @@
+# -*- 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.storage
+
+
+class TestVMVolume(qubesmgmt.tests.QubesTestCase):
+    def setUp(self):
+        super(TestVMVolume, self).setUp()
+        self.vol = qubesmgmt.storage.Volume(self.app, vm='test-vm',
+            vm_name='volname')
+        self.pool_vol = qubesmgmt.storage.Volume(self.app, pool='test-pool',
+            vid='some-id')
+
+    def expect_info(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.Info', 'volname', None)] = \
+            b'0\x00' \
+            b'pool=test-pool\n' \
+            b'vid=some-id\n' \
+            b'size=1024\n' \
+            b'usage=512\n' \
+            b'rw=True\n' \
+            b'snap_on_start=True\n' \
+            b'save_on_stop=True\n' \
+            b'source=\n' \
+            b'internal=True\n' \
+            b'revisions_to_keep=3\n'
+
+    def test_000_qubesd_call(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.TestMethod', 'volname', None)] = \
+            b'0\x00method_result'
+        self.assertEqual(self.vol._qubesd_call('TestMethod'),
+            b'method_result')
+        self.assertAllCalled()
+
+    def test_001_fetch_info(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.Info', 'volname', None)] = \
+            b'0\x00prop1=val1\nprop2=val2\n'
+        self.vol._fetch_info()
+        self.assertEqual(self.vol._info, {'prop1': 'val1', 'prop2': 'val2'})
+        self.assertAllCalled()
+
+    def test_010_pool(self):
+        self.expect_info()
+        self.assertEqual(self.vol.pool, 'test-pool')
+        self.assertAllCalled()
+
+    def test_011_vid(self):
+        self.expect_info()
+        self.assertEqual(self.vol.vid, 'some-id')
+        self.assertAllCalled()
+
+    def test_012_size(self):
+        self.expect_info()
+        self.assertEqual(self.vol.size, 1024)
+        self.assertAllCalled()
+
+    def test_013_usage(self):
+        self.expect_info()
+        self.assertEqual(self.vol.usage, 512)
+        self.assertAllCalled()
+
+    def test_014_rw(self):
+        self.expect_info()
+        self.assertEqual(self.vol.rw, True)
+        self.assertAllCalled()
+
+    def test_015_snap_on_start(self):
+        self.expect_info()
+        self.assertEqual(self.vol.snap_on_start, True)
+        self.assertAllCalled()
+
+    def test_016_save_on_stop(self):
+        self.expect_info()
+        self.assertEqual(self.vol.save_on_stop, True)
+        self.assertAllCalled()
+
+    def test_017_source_none(self):
+        self.expect_info()
+        self.assertEqual(self.vol.source, None)
+        self.assertAllCalled()
+
+    def test_018_source(self):
+        self.expect_info()
+        call_key = list(self.app.expected_calls)[0]
+        self.app.expected_calls[call_key] = self.app.expected_calls[
+            call_key].replace(b'source=\n', b'source=test-pool:other-id\n')
+        self.assertEqual(self.vol.source, 'test-pool:other-id')
+        self.assertAllCalled()
+
+    def test_019_internal(self):
+        self.expect_info()
+        self.assertEqual(self.vol.internal, True)
+        self.assertAllCalled()
+
+    def test_020_revisions_to_keep(self):
+        self.expect_info()
+        self.assertEqual(self.vol.revisions_to_keep, 3)
+        self.assertAllCalled()
+
+    def test_021_revisions(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.ListSnapshots', 'volname', None)] = \
+            b'0\x00' \
+            b'snapid1\n' \
+            b'snapid2\n' \
+            b'snapid3\n'
+        self.assertEqual(self.vol.revisions,
+            ['snapid1', 'snapid2', 'snapid3'])
+        self.assertAllCalled()
+
+    def test_022_revisions_empty(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.ListSnapshots', 'volname', None)] = \
+            b'0\x00'
+        self.assertEqual(self.vol.revisions, [])
+        self.assertAllCalled()
+
+    def test_030_resize(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.Resize', 'volname', b'2048')] = b'0\x00'
+        self.vol.resize(2048)
+        self.assertAllCalled()
+
+    def test_031_revert(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.Revert', 'volname', b'snapid1')] = \
+            b'0\x00'
+        self.vol.revert('snapid1')
+        self.assertAllCalled()
+
+
+class TestPoolVolume(TestVMVolume):
+    def setUp(self):
+        super(TestPoolVolume, self).setUp()
+        self.vol = qubesmgmt.storage.Volume(self.app, pool='test-pool',
+            vid='some-id')
+
+    def test_000_qubesd_call(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.TestMethod',
+            'test-pool:some-id', None)] = \
+            b'0\x00method_result'
+        self.assertEqual(self.vol._qubesd_call('TestMethod'),
+            b'method_result')
+        self.assertAllCalled()
+
+    def expect_info(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.Info', 'test-pool:some-id', None)] = \
+            b'0\x00' \
+            b'pool=test-pool\n' \
+            b'vid=some-id\n' \
+            b'size=1024\n' \
+            b'usage=512\n' \
+            b'rw=True\n' \
+            b'snap_on_start=True\n' \
+            b'save_on_stop=True\n' \
+            b'source=\n' \
+            b'internal=True\n' \
+            b'revisions_to_keep=3\n'
+
+    def test_001_fetch_info(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.Info', 'test-pool:some-id',
+            None)] = \
+            b'0\x00prop1=val1\nprop2=val2\n'
+        self.vol._fetch_info()
+        self.assertEqual(self.vol._info, {'prop1': 'val1', 'prop2': 'val2'})
+        self.assertAllCalled()
+
+    def test_010_pool(self):
+        # this should _not_ produce any api call, as pool is already known
+        self.assertEqual(self.vol.pool, 'test-pool')
+        self.assertAllCalled()
+
+    def test_011_vid(self):
+        # this should _not_ produce any api call, as vid is already known
+        self.assertEqual(self.vol.vid, 'some-id')
+        self.assertAllCalled()
+
+    def test_021_revisions(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.ListSnapshots',
+             'test-pool:some-id', None)] = \
+            b'0\x00' \
+            b'snapid1\n' \
+            b'snapid2\n' \
+            b'snapid3\n'
+        self.assertEqual(self.vol.revisions,
+            ['snapid1', 'snapid2', 'snapid3'])
+        self.assertAllCalled()
+
+    def test_022_revisions_empty(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.ListSnapshots',
+            'test-pool:some-id', None)] = b'0\x00'
+        self.assertEqual(self.vol.revisions, [])
+        self.assertAllCalled()
+
+    def test_030_resize(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.Resize',
+            'test-pool:some-id', b'2048')] = b'0\x00'
+        self.vol.resize(2048)
+        self.assertAllCalled()
+
+    def test_031_revert(self):
+        self.app.expected_calls[
+            ('dom0', 'mgmt.pool.volume.Revert', 'test-pool:some-id',
+            b'snapid1')] = b'0\x00'
+        self.vol.revert('snapid1')
+        self.assertAllCalled()

qubesmgmt/tests/vm/storage.py

@@ -0,0 +1,46 @@
+# -*- 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.vm
+
+
+class TestVMVolumes(qubesmgmt.tests.vm.VMTestCase):
+    def test_000_list_volumes(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\nvolatile\nmodules\n'
+        self.assertEqual(set(self.vm.volumes.keys()),
+            set(['root', 'private', 'volatile', 'modules']))
+        self.assertAllCalled()
+
+    def test_001_volume_get(self):
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.List', None, None)] = \
+            b'0\x00root\nprivate\nvolatile\nmodules\n'
+        vol = self.vm.volumes['private']
+        self.assertEqual(vol._vm, 'test-vm')
+        self.assertEqual(vol._vm_name, 'private')
+        # add it here, to raise exception if was called earlier
+        self.app.expected_calls[
+            ('test-vm', 'mgmt.vm.volume.Info', 'private', None)] = \
+            b'0\x00pool=test-pool\nvid=some-id\n'
+        self.assertEqual(vol.pool, 'test-pool')
+        self.assertEqual(vol.vid, 'some-id')
+        self.assertAllCalled()

qubesmgmt/vm/__init__.py

@@ -21,12 +21,14 @@
 '''Qubes VM objects.'''
 
 import qubesmgmt.base
+import qubesmgmt.storage
 
 
 class QubesVM(qubesmgmt.base.PropertyHolder):
     '''Qubes domain.'''
     def __init__(self, app, name):
         super(QubesVM, self).__init__(app, 'mgmt.vm.property.', name)
+        self._volumes = None
 
     @property
     def name(self):
@@ -41,6 +43,7 @@ class QubesVM(qubesmgmt.base.PropertyHolder):
             'name',
             str(new_value).encode('utf-8'))
         self._method_dest = new_value
+        self._volumes = None
         self.app.domains.clear_cache()
 
     def start(self):
@@ -112,6 +115,19 @@ class QubesVM(qubesmgmt.base.PropertyHolder):
         raise NotImplementedError
         #self.qubesd_call(self._method_dest, 'mgmt.vm.Resume')
 
+    @property
+    def volumes(self):
+        '''VM disk volumes'''
+        if self._volumes is None:
+            volumes_list = self.qubesd_call(
+                self._method_dest, 'mgmt.vm.volume.List')
+            self._volumes = {}
+            for volname in volumes_list.decode('ascii').splitlines():
+                if not volname:
+                    continue
+                self._volumes[volname] = qubesmgmt.storage.Volume(self.app,
+                    vm=self.name, vm_name=volname)
+        return self._volumes
 
 class AdminVM(QubesVM):
     '''Dom0'''