storage: centralize shrinking check from qvm-volume to module

qubesadmin/storage.py

@@ -116,6 +116,11 @@ def __lt__(self, other: object) -> bool:
                 return (self._pool, self._vid) < (other._pool, other._vid)
         return NotImplemented
 
+    @property
+    def vm(self) -> str | None:
+        """Name of the VM the volume belongs to, if available"""
+        return self._vm
+
     @property
     def name(self) -> str | None:
         """per-VM volume name, if available"""
@@ -259,13 +264,24 @@ def is_outdated(self) -> bool:
         assert self._info is not None
         return self._info.get('is_outdated', False) == 'True'
 
-    def resize(self, size: object) -> None:
+    def resize(self, size: int, *, allow_shrink: bool=False) -> None:
         """Resize volume.
 
-        Currently only extending is supported.
-
         :param int size: new size in bytes.
         """
+
+        if not allow_shrink:
+            current_size = self.size
+            if size < current_size:
+                vol_str = self.vm + ":" + self.name
+                raise qubesadmin.exc.StoragePoolException(
+                    f"Refusing to shrink volume {vol_str}"
+                    f" from {current_size} to {size} bytes.\n"
+                    f"If you really know what you are doing:"
+                    f" Manually shrink the filesystem on the volume first,"
+                    f" and repartition the volume if it is partitioned."
+                    f" Do this in a VM, not in dom0."
+                    f" Then use 'qvm-volume resize --force {vol_str} {size}'")
         self._qubesd_call('Resize', str(size).encode('ascii'))
 
     @property

qubesadmin/tests/storage.py

@@ -31,8 +31,6 @@ def setUp(self):
         super().setUp()
         self.vol = qubesadmin.storage.Volume(self.app, vm='test-vm',
             vm_name='volname')
-        self.pool_vol = qubesadmin.storage.Volume(self.app, pool='test-pool',
-            vid='some-id')
 
     def expect_info(self):
         self.app.expected_calls[
@@ -136,6 +134,7 @@ def test_022_revisions_empty(self):
         self.assertAllCalled()
 
     def test_030_resize(self):
+        self.expect_info()
         self.app.expected_calls[
             ('test-vm', 'admin.vm.volume.Resize', 'volname', b'2048')] = \
             b'0\x00'
@@ -175,7 +174,7 @@ def test_050_clone(self):
         self.assertAllCalled()
 
 
-class TestPoolVolume(TestVMVolume):
+class TestPoolVolume(qubesadmin.tests.QubesTestCase):
     def setUp(self):
         super().setUp()
         self.vol = qubesadmin.storage.Volume(self.app, pool='test-pool',
@@ -190,29 +189,6 @@ def test_000_qubesd_call(self):
             b'method_result')
         self.assertAllCalled()
 
-    def expect_info(self):
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.Info', 'test-pool', b'some-id')] = \
-            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'revisions_to_keep=3\n'
-
-    def test_001_fetch_info(self):
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.Info', 'test-pool',
-            b'some-id')] = \
-            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')
@@ -223,57 +199,6 @@ def test_011_vid(self):
         self.assertEqual(self.vol.vid, 'some-id')
         self.assertAllCalled()
 
-    def test_021_revisions(self):
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.ListSnapshots',
-             'test-pool', b'some-id')] = \
-            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', 'admin.pool.volume.ListSnapshots',
-            'test-pool', b'some-id')] = b'0\x00'
-        self.assertEqual(self.vol.revisions, [])
-        self.assertAllCalled()
-
-    def test_030_resize(self):
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.Resize',
-            'test-pool', b'some-id 2048')] = b'0\x00'
-        self.vol.resize(2048)
-        self.assertAllCalled()
-
-    def test_031_revert(self):
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.Revert', 'test-pool',
-            b'some-id snapid1')] = b'0\x00'
-        self.vol.revert('snapid1')
-        self.assertAllCalled()
-
-    def test_040_import_data(self):
-        self.skipTest('admin.pool.volume.Import not supported')
-
-    def test_050_clone(self):
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.CloneFrom', 'test-pool',
-            b'volid')] = b'0\x00abcdef'
-        self.app.expected_calls[
-            ('dom0', 'admin.pool.volume.CloneTo', 'test-pool',
-            b'some-id abcdef')] = b'0\x00'
-        source_vol = qubesadmin.storage.Volume(self.app, pool='test-pool',
-            vid='volid')
-        self.vol.clone(source_vol)
-        self.assertAllCalled()
-
-    def test_050_clone_wrong_volume(self):
-        self.skipTest('admin.pool.volume.Clone not supported')
-
 
 class TestPool(qubesadmin.tests.QubesTestCase):
     def test_000_list(self):

qubesadmin/tests/tools/qvm_volume.py

@@ -232,7 +232,8 @@ def test_012_extend_deny_shrink(self):
                 qubesadmin.tools.qvm_volume.main(
                     ['resize', 'testvm:private', '1GiB'],
                     app=self.app))
-        self.assertIn('shrinking of private is disabled', stderr.getvalue())
+        self.assertIn('Refusing to shrink volume testvm:private',
+            stderr.getvalue())
         self.assertAllCalled()
 
     def test_013_resize_force_shrink(self):

qubesadmin/tools/qvm_volume.py

@@ -242,20 +242,13 @@ def revert_volume(args):
     volume.revert(revision)
 
 
-def extend_volumes(args):
-    """ Called by the parser to execute the :program:`qvm-volume extend`
+def resize_volume(args):
+    """ Called by the parser to execute the :program:`qvm-volume resize`
         subcommand
     """
     volume = args.volume
     size = qubesadmin.utils.parse_size(args.size)
-    if not args.force and size < volume.size:
-        raise qubesadmin.exc.StoragePoolException(
-            'For your own safety, shrinking of %s is'
-            ' disabled (%d < %d). If you really know what you'
-            ' are doing, resize filesystem manually first, then use `-f` '
-            'option.' %
-            (volume.name, size, volume.size))
-    volume.resize(size)
+    volume.resize(size, allow_shrink=args.force)
 
 
 def init_list_parser(sub_parsers):
@@ -291,18 +284,18 @@ def init_revert_parser(sub_parsers):
     revert_parser.set_defaults(func=revert_volume)
 
 
-def init_extend_parser(sub_parsers):
-    """ Add 'extend' action related options """
-    extend_parser = sub_parsers.add_parser(
+def init_resize_parser(sub_parsers):
+    """ Add 'resize' action related options """
+    resize_parser = sub_parsers.add_parser(
         "resize", aliases=('extend', ), help="resize volume for domain")
-    extend_parser.add_argument(metavar='VM:VOLUME', dest='volume',
+    resize_parser.add_argument(metavar='VM:VOLUME', dest='volume',
                                action=qubesadmin.tools.VMVolumeAction)
-    extend_parser.add_argument('size', help='New size in bytes')
-    extend_parser.add_argument(
+    resize_parser.add_argument('size', help='New size in bytes')
+    resize_parser.add_argument(
         '--force', '-f', action='store_true',
         help='Force operation, even if new size is smaller than the current '
              'one')
-    extend_parser.set_defaults(func=extend_volumes)
+    resize_parser.set_defaults(func=resize_volume)
 
 
 def init_info_parser(sub_parsers):
@@ -371,7 +364,7 @@ def get_parser():
         dest='command')
     init_info_parser(sub_parsers)
     init_config_parser(sub_parsers)
-    init_extend_parser(sub_parsers)
+    init_resize_parser(sub_parsers)
     init_list_parser(sub_parsers)
     init_revert_parser(sub_parsers)
     init_import_parser(sub_parsers)