Merge "V2: disallow image format update for active status"

glance/domain/__init__.py

@@ -110,8 +110,8 @@ class Image(object):
         self.locations = kwargs.pop('locations', [])
         self.checksum = kwargs.pop('checksum', None)
         self.owner = kwargs.pop('owner', None)
-        self.disk_format = kwargs.pop('disk_format', None)
-        self.container_format = kwargs.pop('container_format', None)
+        self._disk_format = kwargs.pop('disk_format', None)
+        self._container_format = kwargs.pop('container_format', None)
         self.size = kwargs.pop('size', None)
         extra_properties = kwargs.pop('extra_properties', None) or {}
         self.extra_properties = ExtraProperties(extra_properties)
@@ -157,6 +157,30 @@ class Image(object):
     def tags(self, value):
         self._tags = set(value)
 
+    @property
+    def container_format(self):
+        return self._container_format
+
+    @container_format.setter
+    def container_format(self, value):
+        if hasattr(self, '_container_format') and self.status != 'queued':
+            msg = _("Attribute container_format can be only replaced "
+                    "for a queued image.")
+            raise exception.Forbidden(message=msg)
+        self._container_format = value
+
+    @property
+    def disk_format(self):
+        return self._disk_format
+
+    @disk_format.setter
+    def disk_format(self, value):
+        if hasattr(self, '_disk_format') and self.status != 'queued':
+            msg = _("Attribute disk_format can be only replaced "
+                    "for a queued image.")
+            raise exception.Forbidden(message=msg)
+        self._disk_format = value
+
     def delete(self):
         if self.protected:
             raise exception.ProtectedImageDelete(image_id=self.image_id)

glance/tests/functional/v2/test_images.py

@@ -244,6 +244,7 @@ class TestImages(functional.FunctionalTest):
         data = json.dumps([
             {'op': 'replace', 'path': '/name', 'value': 'image-2'},
             {'op': 'replace', 'path': '/disk_format', 'value': 'vhd'},
+            {'op': 'replace', 'path': '/container_format', 'value': 'ami'},
             {'op': 'replace', 'path': '/foo', 'value': 'baz'},
             {'op': 'add', 'path': '/ping', 'value': 'pong'},
             {'op': 'replace', 'path': '/protected', 'value': True},
@@ -319,6 +320,19 @@ class TestImages(functional.FunctionalTest):
         self.assertEqual('8f113e38d28a79a5a451b16048cc2b72', image['checksum'])
         self.assertEqual('active', image['status'])
 
+        # `disk_format` and `container_format` cannot
+        # be replaced when the image is active.
+        immutable_paths = ['/disk_format', '/container_format']
+        media_type = 'application/openstack-images-v2.1-json-patch'
+        headers = self._headers({'content-type': media_type})
+        path = self._url('/v2/images/%s' % image_id)
+        for immutable_path in immutable_paths:
+            data = json.dumps([
+                {'op': 'replace', 'path': immutable_path, 'value': 'ari'},
+            ])
+            response = requests.patch(path, headers=headers, data=data)
+            self.assertEqual(403, response.status_code)
+
         # Try to download the data that was just uploaded
         path = self._url('/v2/images/%s/file' % image_id)
         headers = self._headers()

glance/tests/unit/v2/test_images_resource.py

@@ -728,6 +728,41 @@ class TestImagesController(base.IsolatedUnitTest):
         self.assertEqual(output.image_id, UUID1)
         self.assertNotEqual(output.created_at, output.updated_at)
 
+    def test_update_format_properties(self):
+        statuses_for_immutability = ['active', 'saving', 'killed']
+        request = unit_test_utils.get_fake_request(is_admin=True)
+        for status in statuses_for_immutability:
+            image = {
+                'id': uuidutils.generate_uuid(),
+                'status': status,
+                'disk_format': 'ari',
+                'container_format': 'ari',
+            }
+            self.db.image_create(None, image)
+            changes = [
+                {'op': 'replace', 'path': ['disk_format'], 'value': 'ami'},
+            ]
+            self.assertRaises(webob.exc.HTTPForbidden,
+                              self.controller.update,
+                              request, image['id'], changes)
+            changes = [
+                {'op': 'replace',
+                 'path': ['container_format'],
+                 'value': 'ami'},
+            ]
+            self.assertRaises(webob.exc.HTTPForbidden,
+                              self.controller.update,
+                              request, image['id'], changes)
+        self.db.image_update(None, image['id'], {'status': 'queued'})
+
+        changes = [
+            {'op': 'replace', 'path': ['disk_format'], 'value': 'raw'},
+            {'op': 'replace', 'path': ['container_format'], 'value': 'bare'},
+        ]
+        resp = self.controller.update(request, image['id'], changes)
+        self.assertEqual(resp.disk_format, 'raw')
+        self.assertEqual(resp.container_format, 'bare')
+
     def test_update_remove_property_while_over_limit(self):
         """
         Ensure that image properties can be removed.