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

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 @@ class FilePool(qubes.storage.Pool):
                 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 @@ class TC_01_FileVolumes(qubes.tests.QubesTestCase):
         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 @@ class TC_03_FilePool(qubes.tests.QubesTestCase):
         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"""