storage/file: move revisions_to_keep restrictions to property setter

Do not check for accepted value only in constructor, do that in property
setter. This will allow enforcing the limit regardless of how the value
was set.

This is preparation for dynamic revisions_to_keep change.

QubesOS/qubes-issues#3256

Author: marmarek

qubes/storage/file.py

@@ -54,6 +54,7 @@ class FilePool(qubes.storage.Pool):
     driver = 'file'
 
     def __init__(self, revisions_to_keep=1, dir_path=None, **kwargs):
+        self._revisions_to_keep = 0
         super(FilePool, self).__init__(revisions_to_keep=revisions_to_keep,
                                        **kwargs)
         assert dir_path, "No pool dir_path specified"
@@ -85,19 +86,27 @@ def init_volume(self, vm, volume_config):
                 volume_config['revisions_to_keep'] = 0
         except KeyError:
             pass
-        finally:
-            if 'revisions_to_keep' not in volume_config:
-                volume_config['revisions_to_keep'] = self.revisions_to_keep
 
-        if int(volume_config['revisions_to_keep']) > 1:
-            raise NotImplementedError(
-                'FilePool supports maximum 1 volume revision to keep')
+        if 'revisions_to_keep' not in volume_config:
+            volume_config['revisions_to_keep'] = self.revisions_to_keep
 
         volume_config['pool'] = self
         volume = FileVolume(**volume_config)
         self._volumes += [volume]
         return volume
 
+    @property
+    def revisions_to_keep(self):
+        return self._revisions_to_keep
+
+    @revisions_to_keep.setter
+    def revisions_to_keep(self, value):
+        value = int(value)
+        if value > 1:
+            raise NotImplementedError(
+                'FilePool supports maximum 1 volume revision to keep')
+        self._revisions_to_keep = value
+
     def destroy(self):
         pass
 
@@ -162,12 +171,24 @@ class FileVolume(qubes.storage.Volume):
     def __init__(self, dir_path, **kwargs):
         self.dir_path = dir_path
         assert self.dir_path, "dir_path not specified"
+        self._revisions_to_keep = 0
         super(FileVolume, self).__init__(**kwargs)
 
         if self.snap_on_start:
             img_name = self.source.vid + '-cow.img'
             self.path_source_cow = os.path.join(self.dir_path, img_name)
 
+    @property
+    def revisions_to_keep(self):
+        return self._revisions_to_keep
+
+    @revisions_to_keep.setter
+    def revisions_to_keep(self, value):
+        if int(value) > 1:
+            raise NotImplementedError(
+                'FileVolume supports maximum 1 volume revision to keep')
+        self._revisions_to_keep = int(value)
+
     def create(self):
         assert isinstance(self.size, int) and self.size > 0, \
             'Volume size must be > 0'

qubes/tests/storage_file.py

@@ -296,6 +296,22 @@ def test_006_template_volumes(self):
         expected = vm_dir + '/volatile.img'
         self.assertVolumePath(vm, 'volatile', expected, rw=True)
 
+    def test_010_revisions_to_keep_reject_invalid(self):
+        ''' Check if TemplateVM volumes are propertly initialized '''
+        config = {
+            'name': 'root',
+            'pool': self.POOL_NAME,
+            'save_on_stop': True,
+            'rw': True,
+            'size': defaults['root_img_size'],
+        }
+        vm = qubes.tests.storage.TestVM(self)
+        volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
+        self.assertEqual(volume.revisions_to_keep, 1)
+        with self.assertRaises((NotImplementedError, ValueError)):
+            volume.revisions_to_keep = 2
+        self.assertEqual(volume.revisions_to_keep, 1)
+
     def assertVolumePath(self, vm, dev_name, expected, rw=True):
         # :pylint: disable=invalid-name
         volumes = vm.volumes
@@ -384,6 +400,13 @@ def test_004_usage(self):
         self.assertEqual(usage,
             statvfs.f_frsize * (statvfs.f_blocks - statvfs.f_bfree))
 
+    def test_005_revisions_to_keep(self):
+        pool = self.app.get_pool(self.POOL_NAME)
+        self.assertEqual(pool.revisions_to_keep, 1)
+        with self.assertRaises((NotImplementedError, ValueError)):
+            pool.revisions_to_keep = 2
+        self.assertEqual(pool.revisions_to_keep, 1)
+
     def test_011_appvm_file_images(self):
         """ Check if all the needed image files are created for an AppVm"""