openstack
/
glance
Code Issues Proposed changes

Add ability to deactivate an image 

This patch provides the ability to 'deactivate' an image by
providing two new API calls and a new image status 'deactivated'.
Attempting to download a deactivated image will result in a
403 'Forbidden' return code. Also, image locations won't be visible
for deactivated images unless the user is admin.
All other image operations should remain unaffected.

The two new API calls are:
    - POST /images/{image_id}/actions/deactivate
    - POST /images/{image_id}/actions/reactivate

DocImpact
UpgradeImpact

Change-Id: I32b7cc7ce8404457a87c8c05041aa2a30152b930
Implements: bp deactivate-image
This commit is contained in:
Eddie Sheffield 2014-11-03 20:29:41 -05:00 committed by Hemanth Makkapati
parent 15fea34808
commit b000c85b7f
21 changed files with 568 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
doc/source/images_src/image_status_transition.dot
View File

@ -40,6 +40,10 @@ digraph {
"active" -> "queued" [label="remove location*"];
"active" -> "pending_delete" [label="delayed delete"];
"active" -> "deleted" [label="delete"];
"active" -> "deactivated" [label="deactivate"];
"deactivated" -> "active" [label="reactivate"];
"deactivated" -> "deleted" [label="delete"];
"killed" -> "deleted" [label="delete"];

BIN
doc/source/images_src/image_status_transition.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 153 KiB

3
etc/policy.json
View File

@ -30,6 +30,9 @@
"add_task": "",
"modify_task": "",
"deactivate": "",
"reactivate": "",
"get_metadef_namespace": "",
"get_metadef_namespaces":"",
"modify_metadef_namespace":"",

5
glance/api/middleware/cache.py
View File

@ -154,6 +154,11 @@ class CacheFilter(wsgi.Middleware):
return None
method = getattr(self, '_get_%s_image_metadata' % version)
image_metadata = method(request, image_id)
# Deactivated images shall not be served from cache
if image_metadata['status'] == 'deactivated':
return None
try:
self._enforce(request, 'download_image', target=image_metadata)
except exception.Forbidden:

14
glance/api/policy.py
View File

@ -165,6 +165,20 @@ class ImageProxy(glance.domain.proxy.Image):
self.policy.enforce(self.context, 'delete_image', {})
return self.image.delete()
def deactivate(self):
LOG.debug('Attempting deactivate')
target = ImageTarget(self.image)
self.policy.enforce(self.context, 'deactivate', target=target)
LOG.debug('Deactivate allowed, continue')
self.image.deactivate()
def reactivate(self):
LOG.debug('Attempting reactivate')
target = ImageTarget(self.image)
self.policy.enforce(self.context, 'reactivate', target=target)
LOG.debug('Reactivate allowed, continue')
self.image.reactivate()
def get_data(self, *args, **kwargs):
target = ImageTarget(self.image)
self.policy.enforce(self.context, 'download_image',

13
glance/api/v1/controller.py
View File

@ -52,15 +52,22 @@ class BaseController(object):
request=request,
content_type='text/plain')
def get_active_image_meta_or_404(self, request, image_id):
def get_active_image_meta_or_error(self, request, image_id):
"""
Same as get_image_meta_or_404 except that it will raise a 404 if the
image isn't 'active'.
Same as get_image_meta_or_404 except that it will raise a 403 if the
image is deactivated or 404 if the image is otherwise not 'active'.
"""
image = self.get_image_meta_or_404(request, image_id)
if image['status'] == 'deactivated':
msg = "Image %s is deactivated" % image_id
LOG.debug(msg)
msg = _("Image %s is deactivated") % image_id
raise webob.exc.HTTPForbidden(
msg, request=request, content_type='type/plain')
if image['status'] != 'active':
msg = "Image %s is not active" % image_id
LOG.debug(msg)
msg = _("Image %s is not active") % image_id
raise webob.exc.HTTPNotFound(
msg, request=request, content_type='text/plain')
return image

2
glance/api/v1/images.py
View File

@ -476,7 +476,7 @@ class Controller(controller.BaseController):
self._enforce(req, 'get_image')
try:
image_meta = self.get_active_image_meta_or_404(req, id)
image_meta = self.get_active_image_meta_or_error(req, id)
except HTTPNotFound:
# provision for backward-compatibility breaking issue
# catch the 404 exception and raise it after enforcing

89
glance/api/v2/image_actions.py Normal file
View File

@ -0,0 +1,89 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import glance_store
from oslo_log import log as logging
import webob.exc
from glance.api import policy
from glance.common import exception
from glance.common import utils
from glance.common import wsgi
import glance.db
import glance.gateway
from glance import i18n
import glance.notifier
LOG = logging.getLogger(__name__)
_ = i18n._
_LI = i18n._LI
class ImageActionsController(object):
def __init__(self, db_api=None, policy_enforcer=None, notifier=None,
store_api=None):
self.db_api = db_api or glance.db.get_api()
self.policy = policy_enforcer or policy.Enforcer()
self.notifier = notifier or glance.notifier.Notifier()
self.store_api = store_api or glance_store
self.gateway = glance.gateway.Gateway(self.db_api, self.store_api,
self.notifier, self.policy)
@utils.mutating
def deactivate(self, req, image_id):
image_repo = self.gateway.get_repo(req.context)
try:
image = image_repo.get(image_id)
image.deactivate()
image_repo.save(image)
LOG.info(_LI("Image %s is deactivated") % image_id)
except exception.NotFound as e:
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Forbidden as e:
raise webob.exc.HTTPForbidden(explanation=e.msg)
except exception.InvalidImageStatusTransition as e:
raise webob.exc.HTTPBadRequest(explanation=e.msg)
@utils.mutating
def reactivate(self, req, image_id):
image_repo = self.gateway.get_repo(req.context)
try:
image = image_repo.get(image_id)
image.reactivate()
image_repo.save(image)
LOG.info(_LI("Image %s is reactivated") % image_id)
except exception.NotFound as e:
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Forbidden as e:
raise webob.exc.HTTPForbidden(explanation=e.msg)
except exception.InvalidImageStatusTransition as e:
raise webob.exc.HTTPBadRequest(explanation=e.msg)
class ResponseSerializer(wsgi.JSONResponseSerializer):
def deactivate(self, response, result):
response.status_int = 204
def reactivate(self, response, result):
response.status_int = 204
def create_resource():
"""Image data resource factory method"""
deserializer = None
serializer = ResponseSerializer()
controller = ImageActionsController()
return wsgi.Resource(controller, deserializer, serializer)

4
glance/api/v2/image_data.py
View File

@ -169,6 +169,10 @@ class ImageDataController(object):
image_repo = self.gateway.get_repo(req.context)
try:
image = image_repo.get(image_id)
if image.status == 'deactivated':
msg = _('The requested image has been deactivated. '
'Image data download is forbidden.')
raise exception.Forbidden(message=msg)
if not image.locations:
raise exception.ImageDataNotFound()
except exception.ImageDataNotFound as e:

23
glance/api/v2/router.py
View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from glance.api.v2 import image_actions
from glance.api.v2 import image_data
from glance.api.v2 import image_members
from glance.api.v2 import image_tags
@ -447,6 +448,28 @@ class API(wsgi.Router):
allowed_methods='GET, PATCH, DELETE',
conditions={'method': ['POST', 'PUT', 'HEAD']})
image_actions_resource = image_actions.create_resource()
mapper.connect('/images/{image_id}/actions/deactivate',
controller=image_actions_resource,
action='deactivate',
conditions={'method': ['POST']})
mapper.connect('/images/{image_id}/actions/reactivate',
controller=image_actions_resource,
action='reactivate',
conditions={'method': ['POST']})
mapper.connect('/images/{image_id}/actions/deactivate',
controller=reject_method_resource,
action='reject',
allowed_methods='POST',
conditions={'method': ['GET', 'PUT', 'DELETE', 'PATCH',
'HEAD']})
mapper.connect('/images/{image_id}/actions/reactivate',
controller=reject_method_resource,
action='reject',
allowed_methods='POST',
conditions={'method': ['GET', 'PUT', 'DELETE', 'PATCH',
'HEAD']})
image_data_resource = image_data.create_resource()
mapper.connect('/images/{image_id}/file',
controller=image_data_resource,

18
glance/db/simple/api.py
View File

@ -395,7 +395,7 @@ def _image_get(context, image_id, force_show_deleted=False, status=None):
@log_call
def image_get(context, image_id, session=None, force_show_deleted=False):
image = _image_get(context, image_id, force_show_deleted)
return _normalize_locations(copy.deepcopy(image),
return _normalize_locations(context, copy.deepcopy(image),
force_show_deleted=force_show_deleted)
@ -415,7 +415,7 @@ def image_get_all(context, filters=None, marker=None, limit=None,
force_show_deleted = True if filters.get('deleted') else False
res = []
for image in images:
img = _normalize_locations(copy.deepcopy(image),
img = _normalize_locations(context, copy.deepcopy(image),
force_show_deleted=force_show_deleted)
if return_tag:
img['tags'] = image_tag_get_all(context, img['id'])
@ -622,7 +622,7 @@ def _image_locations_delete_all(context, image_id, delete_time=None):
del DATA['locations'][i]
def _normalize_locations(image, force_show_deleted=False):
def _normalize_locations(context, image, force_show_deleted=False):
"""
Generate suitable dictionary list for locations field of image.
@ -630,6 +630,11 @@ def _normalize_locations(image, force_show_deleted=False):
from image query.
"""
if image['status'] == 'deactivated' and not context.is_admin:
# Locations are not returned for a deactivated image for non-admin user
image['locations'] = []
return image
if force_show_deleted:
locations = image['locations']
else:
@ -668,7 +673,7 @@ def image_create(context, image_values):
DATA['images'][image_id] = image
DATA['tags'][image_id] = image.pop('tags', [])
return _normalize_locations(copy.deepcopy(image))
return _normalize_locations(context, copy.deepcopy(image))
@log_call
@ -696,7 +701,7 @@ def image_update(context, image_id, image_values, purge_props=False,
image['updated_at'] = timeutils.utcnow()
_image_update(image, image_values, new_properties)
DATA['images'][image_id] = image
return _normalize_locations(copy.deepcopy(image))
return _normalize_locations(context, copy.deepcopy(image))
@log_call
@ -727,7 +732,8 @@ def image_destroy(context, image_id):
for tag in tags:
image_tag_delete(context, image_id, tag)
return _normalize_locations(copy.deepcopy(DATA['images'][image_id]))
return _normalize_locations(context,
copy.deepcopy(DATA['images'][image_id]))
except KeyError:
raise exception.NotFound()

15
glance/db/sqlalchemy/api.py
View File

@ -58,7 +58,7 @@ _LW = i18n._LW
STATUSES = ['active', 'saving', 'queued', 'killed', 'pending_delete',
'deleted']
'deleted', 'deactivated']
CONF = cfg.CONF
CONF.import_group("profiler", "glance.common.wsgi")
@ -159,10 +159,10 @@ def image_destroy(context, image_id):
_image_tag_delete_all(context, image_id, delete_time, session)
return _normalize_locations(image_ref)
return _normalize_locations(context, image_ref)
def _normalize_locations(image, force_show_deleted=False):
def _normalize_locations(context, image, force_show_deleted=False):
"""
Generate suitable dictionary list for locations field of image.
@ -170,6 +170,11 @@ def _normalize_locations(image, force_show_deleted=False):
from image query.
"""
if image['status'] == 'deactivated' and not context.is_admin:
# Locations are not returned for a deactivated image for non-admin user
image['locations'] = []
return image
if force_show_deleted:
locations = image['locations']
else:
@ -191,7 +196,7 @@ def _normalize_tags(image):
def image_get(context, image_id, session=None, force_show_deleted=False):
image = _image_get(context, image_id, session=session,
force_show_deleted=force_show_deleted)
image = _normalize_locations(image.to_dict(),
image = _normalize_locations(context, image.to_dict(),
force_show_deleted=force_show_deleted)
return image
@ -616,7 +621,7 @@ def image_get_all(context, filters=None, marker=None, limit=None,
images = []
for image in query.all():
image_dict = image.to_dict()
image_dict = _normalize_locations(image_dict,
image_dict = _normalize_locations(context, image_dict,
force_show_deleted=showing_deleted)
if return_tag:
image_dict = _normalize_tags(image_dict)

31
glance/domain/__init__.py
View File

@ -107,10 +107,11 @@ class Image(object):
# can be retried.
'queued': ('saving', 'active', 'deleted'),
'saving': ('active', 'killed', 'deleted', 'queued'),
'active': ('queued', 'pending_delete', 'deleted'),
'active': ('queued', 'pending_delete', 'deleted', 'deactivated'),
'killed': ('deleted',),
'pending_delete': ('deleted',),
'deleted': (),
'deactivated': ('active', 'deleted'),
}
def __init__(self, image_id, status, created_at, updated_at, **kwargs):
@ -246,6 +247,34 @@ class Image(object):
else:
self.status = 'deleted'
def deactivate(self):
if self.status == 'active':
self.status = 'deactivated'
elif self.status == 'deactivated':
# Noop if already deactive
pass
else:
msg = ("Not allowed to deactivate image in status '%s'"
% self.status)
LOG.debug(msg)
msg = (_("Not allowed to deactivate image in status '%s'")
% self.status)
raise exception.Forbidden(message=msg)
def reactivate(self):
if self.status == 'deactivated':
self.status = 'active'
elif self.status == 'active':
# Noop if already active
pass
else:
msg = ("Not allowed to reactivate image in status '%s'"
% self.status)
LOG.debug(msg)
msg = (_("Not allowed to reactivate image in status '%s'")
% self.status)
raise exception.Forbidden(message=msg)
def get_data(self, *args, **kwargs):
raise NotImplementedError()

6
glance/domain/proxy.py
View File

@ -156,6 +156,12 @@ class Image(object):
def delete(self):
self.base.delete()
def deactivate(self):
self.base.deactivate()
def reactivate(self):
self.base.reactivate()
def set_data(self, data, size=None):
self.base.set_data(data, size)

5
glance/tests/etc/policy.json
View File

@ -54,5 +54,8 @@
"get_metadef_tags":"",
"modify_metadef_tag":"",
"add_metadef_tag":"",
"add_metadef_tags":""
"add_metadef_tags":"",
"deactivate": "",
"reactivate": ""
}

94
glance/tests/functional/test_cache_middleware.py
View File

@ -332,6 +332,100 @@ class BaseCacheMiddlewareTest(object):
self.stop_servers()
@skip_if_disabled
def test_cache_middleware_trans_with_deactivated_image(self):
"""
Ensure the image v1/v2 API image transfer forbids downloading
deactivated images.
Image deactivation is not available in v1. So, we'll deactivate the
image using v2 but test image transfer with both v1 and v2.
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# Add an image and verify a 200 OK is returned
image_data = "*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(201, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
image_id = data['image']['id']
# Grab the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(200, response.status)
# Verify image in cache
image_cached_path = os.path.join(self.api_server.image_cache_dir,
image_id)
self.assertTrue(os.path.exists(image_cached_path))
# Deactivate the image using v2
path = "http://%s:%d/v2/images/%s/actions/deactivate"
path = path % ("127.0.0.1", self.api_port, image_id)
http = httplib2.Http()
response, content = http.request(path, 'POST')
self.assertEqual(204, response.status)
# Download the image with v1. Ensure it is forbidden
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(403, response.status)
# Download the image with v2. Ensure it is forbidden
path = "http://%s:%d/v2/images/%s/file" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(403, response.status)
# Reactivate the image using v2
path = "http://%s:%d/v2/images/%s/actions/reactivate"
path = path % ("127.0.0.1", self.api_port, image_id)
http = httplib2.Http()
response, content = http.request(path, 'POST')
self.assertEqual(204, response.status)
# Download the image with v1. Ensure it is allowed
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(200, response.status)
# Download the image with v2. Ensure it is allowed
path = "http://%s:%d/v2/images/%s/file" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(200, response.status)
# Now, we delete the image from the server and verify that
# the image cache no longer contains the deleted image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(200, response.status)
self.assertFalse(os.path.exists(image_cached_path))
self.stop_servers()
class BaseCacheManageMiddlewareTest(object):

34
glance/tests/functional/v2/test_images.py
View File

@ -410,6 +410,40 @@ class TestImages(functional.FunctionalTest):
self.assertEqual(200, response.status_code)
self.assertEqual(5, jsonutils.loads(response.text)['size'])
# Should be able to deactivate image
path = self._url('/v2/images/%s/actions/deactivate' % image_id)
response = requests.post(path, data={}, headers=self._headers())
self.assertEqual(204, response.status_code)
# Deactivating a deactivated image succeeds (no-op)
path = self._url('/v2/images/%s/actions/deactivate' % image_id)
response = requests.post(path, data={}, headers=self._headers())
self.assertEqual(204, response.status_code)
# Can't download a deactivated image
path = self._url('/v2/images/%s/file' % image_id)
response = requests.get(path, headers=self._headers())
self.assertEqual(403, response.status_code)
# Deactivated image should still be in a listing
path = self._url('/v2/images')
response = requests.get(path, headers=self._headers())
self.assertEqual(200, response.status_code)
images = jsonutils.loads(response.text)['images']
self.assertEqual(2, len(images))
self.assertEqual(image2_id, images[0]['id'])
self.assertEqual(image_id, images[1]['id'])
# Should be able to reactivate a deactivated image
path = self._url('/v2/images/%s/actions/reactivate' % image_id)
response = requests.post(path, data={}, headers=self._headers())
self.assertEqual(204, response.status_code)
# Reactivating an active image succeeds (no-op)
path = self._url('/v2/images/%s/actions/reactivate' % image_id)
response = requests.post(path, data={}, headers=self._headers())
self.assertEqual(204, response.status_code)
# Deletion should not work on protected images
path = self._url('/v2/images/%s' % image_id)
response = requests.delete(path, headers=self._headers())

4
glance/tests/unit/test_cache_middleware.py
View File

@ -223,7 +223,7 @@ class TestCacheMiddlewareProcessRequest(base.IsolatedUnitTest):
raise exception.NotFound()
def fake_get_v1_image_metadata(request, image_id):
return {'properties': {}}
return {'status': 'active', 'properties': {}}
image_id = 'test1'
request = webob.Request.blank('/v1/images/%s' % image_id)
@ -386,7 +386,7 @@ class TestCacheMiddlewareProcessRequest(base.IsolatedUnitTest):
"""
def fake_get_v1_image_metadata(*args, **kwargs):
return {'properties': {}}
return {'status': 'active', 'properties': {}}
image_id = 'test1'
request = webob.Request.blank('/v1/images/%s' % image_id)

25
glance/tests/unit/v1/test_api.py
View File

@ -52,6 +52,7 @@ _gen_uuid = lambda: str(uuid.uuid4())
UUID1 = _gen_uuid()
UUID2 = _gen_uuid()
UUID3 = _gen_uuid()
class TestGlanceAPI(base.IsolatedUnitTest):
@ -90,6 +91,21 @@ class TestGlanceAPI(base.IsolatedUnitTest):
'size': 19,
'locations': [{'url': "file:///%s/%s" % (self.test_dir, UUID2),
'metadata': {}, 'status': 'active'}],
'properties': {}},
{'id': UUID3,
'name': 'fake image #3',
'status': 'deactivated',
'disk_format': 'ami',
'container_format': 'ami',
'is_public': False,
'created_at': timeutils.utcnow(),
'updated_at': timeutils.utcnow(),
'deleted_at': None,
'deleted': False,
'checksum': None,
'size': 13,
'locations': [{'url': "file:///%s/%s" % (self.test_dir, UUID1),
'metadata': {}, 'status': 'active'}],
'properties': {}}]
self.context = glance.context.RequestContext(is_admin=True)
db_api.get_engine()
@ -1296,6 +1312,13 @@ class TestGlanceAPI(base.IsolatedUnitTest):
"""Tests delayed activation of image with missing container format"""
self._do_test_put_image_content_missing_format('container_format')
def test_download_deactivated_images(self):
"""Tests exception raised trying to download a deactivated image"""
req = webob.Request.blank("/images/%s" % UUID3)
req.method = 'GET'
res = req.get_response(self.api)
self.assertEqual(403, res.status_int)
def test_update_deleted_image(self):
"""Tests that exception raised trying to update a deleted image"""
req = webob.Request.blank("/images/%s" % UUID2)
@ -2615,7 +2638,7 @@ class TestGlanceAPI(base.IsolatedUnitTest):
image_controller = glance.api.v1.images.Controller()
with mock.patch.object(image_controller,
'get_active_image_meta_or_404'
'get_active_image_meta_or_error'
) as mocked_get_image:
mocked_get_image.return_value = image_fixture
self.assertRaises(webob.exc.HTTPServiceUnavailable,

189
glance/tests/unit/v2/test_image_actions_resource.py Normal file
View File

@ -0,0 +1,189 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import glance_store as store
import webob
import glance.api.v2.image_actions as image_actions
import glance.context
from glance.tests.unit import base
import glance.tests.unit.utils as unit_test_utils
BASE_URI = unit_test_utils.BASE_URI
USER1 = '54492ba0-f4df-4e4e-be62-27f4d76b29cf'
UUID1 = 'c80a1a6c-bd1f-41c5-90ee-81afedb1d58d'
TENANT1 = '6838eb7b-6ded-434a-882c-b344c77fe8df'
CHKSUM = '93264c3edf5972c9f1cb309543d38a5c'
def _db_fixture(id, **kwargs):
obj = {
'id': id,
'name': None,
'is_public': False,
'properties': {},
'checksum': None,
'owner': None,
'status': 'queued',
'tags': [],
'size': None,
'virtual_size': None,
'locations': [],
'protected': False,
'disk_format': None,
'container_format': None,
'deleted': False,
'min_ram': None,
'min_disk': None,
}
obj.update(kwargs)
return obj
class TestImageActionsController(base.IsolatedUnitTest):
def setUp(self):
super(TestImageActionsController, self).setUp()
self.db = unit_test_utils.FakeDB()
self.policy = unit_test_utils.FakePolicyEnforcer()
self.notifier = unit_test_utils.FakeNotifier()
self.store = unit_test_utils.FakeStoreAPI()
for i in range(1, 4):
self.store.data['%s/fake_location_%i' % (BASE_URI, i)] = ('Z', 1)
self.store_utils = unit_test_utils.FakeStoreUtils(self.store)
self.controller = image_actions.ImageActionsController(
self.db,
self.policy,
self.notifier,
self.store)
self.controller.gateway.store_utils = self.store_utils
store.create_stores()
def _get_fake_context(self, user=USER1, tenant=TENANT1, roles=['member'],
is_admin=False):
kwargs = {
'user': user,
'tenant': tenant,
'roles': roles,
'is_admin': is_admin,
}
context = glance.context.RequestContext(**kwargs)
return context
def _create_image(self, status):
self.db.reset()
self.images = [
_db_fixture(UUID1, owner=TENANT1, checksum=CHKSUM,
name='1', size=256, virtual_size=1024,
is_public=True,
locations=[{'url': '%s/%s' % (BASE_URI, UUID1),
'metadata': {}, 'status': 'active'}],
disk_format='raw',
container_format='bare',
status=status),
]
context = self._get_fake_context()
[self.db.image_create(context, image) for image in self.images]
def test_deactivate_from_active(self):
self._create_image('active')
request = unit_test_utils.get_fake_request()
self.controller.deactivate(request, UUID1)
image = self.db.image_get(request.context, UUID1)
self.assertEqual('deactivated', image['status'])
def test_deactivate_from_deactivated(self):
self._create_image('deactivated')
request = unit_test_utils.get_fake_request()
self.controller.deactivate(request, UUID1)
image = self.db.image_get(request.context, UUID1)
self.assertEqual('deactivated', image['status'])
def _test_deactivate_from_wrong_status(self, status):
# deactivate will yield an error if the initial status is anything
# other than 'active' or 'deactivated'
self._create_image(status)
request = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPForbidden, self.controller.deactivate,
request, UUID1)
def test_deactivate_from_queued(self):
self._test_deactivate_from_wrong_status('queued')
def test_deactivate_from_saving(self):
self._test_deactivate_from_wrong_status('saving')
def test_deactivate_from_killed(self):
self._test_deactivate_from_wrong_status('killed')
def test_deactivate_from_pending_delete(self):
self._test_deactivate_from_wrong_status('pending_delete')
def test_deactivate_from_deleted(self):
self._test_deactivate_from_wrong_status('deleted')
def test_reactivate_from_active(self):
self._create_image('active')
request = unit_test_utils.get_fake_request()
self.controller.reactivate(request, UUID1)
image = self.db.image_get(request.context, UUID1)
self.assertEqual('active', image['status'])
def test_reactivate_from_deactivated(self):
self._create_image('deactivated')
request = unit_test_utils.get_fake_request()
self.controller.reactivate(request, UUID1)
image = self.db.image_get(request.context, UUID1)
self.assertEqual('active', image['status'])
def _test_reactivate_from_wrong_status(self, status):
# reactivate will yield an error if the initial status is anything
# other than 'active' or 'deactivated'
self._create_image(status)
request = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPForbidden, self.controller.reactivate,
request, UUID1)
def test_reactivate_from_queued(self):
self._test_reactivate_from_wrong_status('queued')
def test_reactivate_from_saving(self):
self._test_reactivate_from_wrong_status('saving')
def test_reactivate_from_killed(self):
self._test_reactivate_from_wrong_status('killed')
def test_reactivate_from_pending_delete(self):
self._test_reactivate_from_wrong_status('pending_delete')
def test_reactivate_from_deleted(self):
self._test_reactivate_from_wrong_status('deleted')

10
glance/tests/unit/v2/test_image_data_resource.py
View File

@ -110,6 +110,16 @@ class TestImagesController(base.StoreClearingUnitTest):
image = self.controller.download(request, unit_test_utils.UUID1)
self.assertEqual('abcd', image.image_id)
def test_download_deactivated(self):
request = unit_test_utils.get_fake_request()
image = FakeImage('abcd',
status='deactivated',
locations=[{'url': 'http://example.com/image',
'metadata': {}, 'status': 'active'}])
self.image_repo.result = image
self.assertRaises(webob.exc.HTTPForbidden, self.controller.download,
request, str(uuid.uuid4()))
def test_download_no_location(self):
request = unit_test_utils.get_fake_request()
self.image_repo.result = FakeImage('abcd')