OpenStack Image Management (Glance)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

3446 lines
149 KiB

# Copyright 2012 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 datetime
import os
import uuid
import glance_store as store
import mock
from oslo.config import cfg
from oslo.serialization import jsonutils
import six
import testtools
import webob
import glance.api.v2.images
from glance.common import exception
from glance import domain
import glance.schema
from glance.tests.unit import base
import glance.tests.unit.utils as unit_test_utils
import glance.tests.utils as test_utils
DATETIME = datetime.datetime(2012, 5, 16, 15, 27, 36, 325355)
ISOTIME = '2012-05-16T15:27:36Z'
CONF = cfg.CONF
BASE_URI = unit_test_utils.BASE_URI
UUID1 = 'c80a1a6c-bd1f-41c5-90ee-81afedb1d58d'
UUID2 = 'a85abd86-55b3-4d5b-b0b4-5d0a6e6042fc'
UUID3 = '971ec09a-8067-4bc8-a91f-ae3557f1c4c7'
UUID4 = '6bbe7cc2-eae7-4c0f-b50d-a7160b0c6a86'
TENANT1 = '6838eb7b-6ded-434a-882c-b344c77fe8df'
TENANT2 = '2c014f32-55eb-467d-8fcb-4bd706012f81'
TENANT3 = '5a3e60e8-cfa9-4a9e-a90a-62b42cea92b8'
TENANT4 = 'c6c87f25-8a94-47ed-8c83-053c25f42df4'
CHKSUM = '93264c3edf5972c9f1cb309543d38a5c'
CHKSUM1 = '43254c3edf6972c9f1cb309543d38a8c'
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
def _domain_fixture(id, **kwargs):
properties = {
'image_id': id,
'name': None,
'visibility': 'private',
'checksum': None,
'owner': None,
'status': 'queued',
'size': None,
'virtual_size': None,
'locations': [],
'protected': False,
'disk_format': None,
'container_format': None,
'min_ram': None,
'min_disk': None,
'tags': [],
}
properties.update(kwargs)
return glance.domain.Image(**properties)
def _db_image_member_fixture(image_id, member_id, **kwargs):
obj = {
'image_id': image_id,
'member': member_id,
}
obj.update(kwargs)
return obj
class TestImagesController(base.IsolatedUnitTest):
def setUp(self):
super(TestImagesController, 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._create_images()
self._create_image_members()
self.controller = glance.api.v2.images.ImagesController(self.db,
self.policy,
self.notifier,
self.store)
self.controller.gateway.store_utils = self.store_utils
store.create_stores()
def _create_images(self):
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='active'),
_db_fixture(UUID2, owner=TENANT1, checksum=CHKSUM1,
name='2', size=512, virtual_size=2048,
is_public=True,
disk_format='raw',
container_format='bare',
status='active',
tags=['redhat', '64bit', 'power'],
properties={'hypervisor_type': 'kvm', 'foo': 'bar',
'bar': 'foo'}),
_db_fixture(UUID3, owner=TENANT3, checksum=CHKSUM1,
name='3', size=512, virtual_size=2048,
is_public=True, tags=['windows', '64bit', 'x86']),
_db_fixture(UUID4, owner=TENANT4, name='4',
size=1024, virtual_size=3072),
]
[self.db.image_create(None, image) for image in self.images]
self.db.image_tag_set_all(None, UUID1, ['ping', 'pong'])
def _create_image_members(self):
self.image_members = [
_db_image_member_fixture(UUID4, TENANT2),
_db_image_member_fixture(UUID4, TENANT3,
status='accepted'),
]
[self.db.image_member_create(None, image_member)
for image_member in self.image_members]
def test_index(self):
self.config(limit_param_default=1, api_limit_max=3)
request = unit_test_utils.get_fake_request()
output = self.controller.index(request)
self.assertEqual(1, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID3])
self.assertEqual(expected, actual)
def test_index_member_status_accepted(self):
self.config(limit_param_default=5, api_limit_max=5)
request = unit_test_utils.get_fake_request(tenant=TENANT2)
output = self.controller.index(request)
self.assertEqual(3, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID1, UUID2, UUID3])
# can see only the public image
self.assertEqual(expected, actual)
request = unit_test_utils.get_fake_request(tenant=TENANT3)
output = self.controller.index(request)
self.assertEqual(4, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID1, UUID2, UUID3, UUID4])
self.assertEqual(expected, actual)
def test_index_admin(self):
request = unit_test_utils.get_fake_request(is_admin=True)
output = self.controller.index(request)
self.assertEqual(4, len(output['images']))
def test_index_admin_deleted_images_hidden(self):
request = unit_test_utils.get_fake_request(is_admin=True)
self.controller.delete(request, UUID1)
output = self.controller.index(request)
self.assertEqual(3, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2, UUID3, UUID4])
self.assertEqual(expected, actual)
def test_index_return_parameters(self):
self.config(limit_param_default=1, api_limit_max=3)
request = unit_test_utils.get_fake_request()
output = self.controller.index(request, marker=UUID3, limit=1,
sort_key=['created_at'],
sort_dir='desc')
self.assertEqual(1, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2])
self.assertEqual(actual, expected)
self.assertEqual(UUID2, output['next_marker'])
def test_index_next_marker(self):
self.config(limit_param_default=1, api_limit_max=3)
request = unit_test_utils.get_fake_request()
output = self.controller.index(request, marker=UUID3, limit=2)
self.assertEqual(2, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2, UUID1])
self.assertEqual(expected, actual)
self.assertEqual(UUID1, output['next_marker'])
def test_index_no_next_marker(self):
self.config(limit_param_default=1, api_limit_max=3)
request = unit_test_utils.get_fake_request()
output = self.controller.index(request, marker=UUID1, limit=2)
self.assertEqual(0, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([])
self.assertEqual(expected, actual)
self.assertNotIn('next_marker', output)
def test_index_with_id_filter(self):
request = unit_test_utils.get_fake_request('/images?id=%s' % UUID1)
output = self.controller.index(request, filters={'id': UUID1})
self.assertEqual(1, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID1])
self.assertEqual(expected, actual)
def test_index_with_checksum_filter_single_image(self):
req = unit_test_utils.get_fake_request('/images?checksum=%s' % CHKSUM)
output = self.controller.index(req, filters={'checksum': CHKSUM})
self.assertEqual(1, len(output['images']))
actual = list([image.image_id for image in output['images']])
expected = [UUID1]
self.assertEqual(expected, actual)
def test_index_with_checksum_filter_multiple_images(self):
req = unit_test_utils.get_fake_request('/images?checksum=%s' % CHKSUM1)
output = self.controller.index(req, filters={'checksum': CHKSUM1})
self.assertEqual(2, len(output['images']))
actual = list([image.image_id for image in output['images']])
expected = [UUID3, UUID2]
self.assertEqual(expected, actual)
def test_index_with_non_existent_checksum(self):
req = unit_test_utils.get_fake_request('/images?checksum=236231827')
output = self.controller.index(req, filters={'checksum': '236231827'})
self.assertEqual(0, len(output['images']))
def test_index_size_max_filter(self):
request = unit_test_utils.get_fake_request('/images?size_max=512')
output = self.controller.index(request, filters={'size_max': 512})
self.assertEqual(3, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID1, UUID2, UUID3])
self.assertEqual(expected, actual)
def test_index_size_min_filter(self):
request = unit_test_utils.get_fake_request('/images?size_min=512')
output = self.controller.index(request, filters={'size_min': 512})
self.assertEqual(2, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2, UUID3])
self.assertEqual(expected, actual)
def test_index_size_range_filter(self):
path = '/images?size_min=512&size_max=512'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'size_min': 512,
'size_max': 512})
self.assertEqual(2, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2, UUID3])
self.assertEqual(expected, actual)
def test_index_virtual_size_max_filter(self):
ref = '/images?virtual_size_max=2048'
request = unit_test_utils.get_fake_request(ref)
output = self.controller.index(request,
filters={'virtual_size_max': 2048})
self.assertEqual(3, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID1, UUID2, UUID3])
self.assertEqual(expected, actual)
def test_index_virtual_size_min_filter(self):
ref = '/images?virtual_size_min=2048'
request = unit_test_utils.get_fake_request(ref)
output = self.controller.index(request,
filters={'virtual_size_min': 2048})
self.assertEqual(2, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2, UUID3])
self.assertEqual(expected, actual)
def test_index_virtual_size_range_filter(self):
path = '/images?virtual_size_min=512&virtual_size_max=2048'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'virtual_size_min': 2048,
'virtual_size_max': 2048})
self.assertEqual(2, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID2, UUID3])
self.assertEqual(expected, actual)
def test_index_with_invalid_max_range_filter_value(self):
request = unit_test_utils.get_fake_request('/images?size_max=blah')
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.index,
request,
filters={'size_max': 'blah'})
def test_index_with_filters_return_many(self):
path = '/images?status=queued'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, filters={'status': 'queued'})
self.assertEqual(1, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID3])
self.assertEqual(expected, actual)
def test_index_with_nonexistent_name_filter(self):
request = unit_test_utils.get_fake_request('/images?name=%s' % 'blah')
images = self.controller.index(request,
filters={'name': 'blah'})['images']
self.assertEqual(0, len(images))
def test_index_with_non_default_is_public_filter(self):
image = _db_fixture(str(uuid.uuid4()),
is_public=False,
owner=TENANT3)
self.db.image_create(None, image)
path = '/images?visibility=private'
request = unit_test_utils.get_fake_request(path, is_admin=True)
output = self.controller.index(request,
filters={'visibility': 'private'})
self.assertEqual(2, len(output['images']))
def test_index_with_many_filters(self):
url = '/images?status=queued&name=3'
request = unit_test_utils.get_fake_request(url)
output = self.controller.index(request,
filters={
'status': 'queued',
'name': '3',
})
self.assertEqual(1, len(output['images']))
actual = set([image.image_id for image in output['images']])
expected = set([UUID3])
self.assertEqual(expected, actual)
def test_index_with_marker(self):
self.config(limit_param_default=1, api_limit_max=3)
path = '/images'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, marker=UUID3)
actual = set([image.image_id for image in output['images']])
self.assertEqual(1, len(actual))
self.assertIn(UUID2, actual)
def test_index_with_limit(self):
path = '/images'
limit = 2
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, limit=limit)
actual = set([image.image_id for image in output['images']])
self.assertEqual(limit, len(actual))
self.assertIn(UUID3, actual)
self.assertIn(UUID2, actual)
def test_index_greater_than_limit_max(self):
self.config(limit_param_default=1, api_limit_max=3)
path = '/images'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, limit=4)
actual = set([image.image_id for image in output['images']])
self.assertEqual(3, len(actual))
self.assertNotIn(output['next_marker'], output)
def test_index_default_limit(self):
self.config(limit_param_default=1, api_limit_max=3)
path = '/images'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request)
actual = set([image.image_id for image in output['images']])
self.assertEqual(1, len(actual))
def test_index_with_sort_dir(self):
path = '/images'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, sort_dir='asc', limit=3)
actual = [image.image_id for image in output['images']]
self.assertEqual(3, len(actual))
self.assertEqual(UUID1, actual[0])
self.assertEqual(UUID2, actual[1])
self.assertEqual(UUID3, actual[2])
def test_index_with_sort_key(self):
path = '/images'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, sort_key=['created_at'],
limit=3)
actual = [image.image_id for image in output['images']]
self.assertEqual(3, len(actual))
self.assertEqual(UUID3, actual[0])
self.assertEqual(UUID2, actual[1])
self.assertEqual(UUID1, actual[2])
def test_index_with_multiple_sort_keys(self):
path = '/images'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
sort_key=['created_at', 'name'],
limit=3)
actual = [image.image_id for image in output['images']]
self.assertEqual(3, len(actual))
self.assertEqual(UUID3, actual[0])
self.assertEqual(UUID2, actual[1])
self.assertEqual(UUID1, actual[2])
def test_index_with_marker_not_found(self):
fake_uuid = str(uuid.uuid4())
path = '/images'
request = unit_test_utils.get_fake_request(path)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.index, request, marker=fake_uuid)
def test_index_invalid_sort_key(self):
path = '/images'
request = unit_test_utils.get_fake_request(path)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.index, request, sort_key=['foo'])
def test_index_zero_images(self):
self.db.reset()
request = unit_test_utils.get_fake_request()
output = self.controller.index(request)
self.assertEqual([], output['images'])
def test_index_with_tags(self):
path = '/images?tag=64bit'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, filters={'tags': ['64bit']})
actual = [image.tags for image in output['images']]
self.assertEqual(2, len(actual))
self.assertIn('64bit', actual[0])
self.assertIn('64bit', actual[1])
def test_index_with_multi_tags(self):
path = '/images?tag=power&tag=64bit'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'tags': ['power', '64bit']})
actual = [image.tags for image in output['images']]
self.assertEqual(1, len(actual))
self.assertIn('64bit', actual[0])
self.assertIn('power', actual[0])
def test_index_with_multi_tags_and_nonexistent(self):
path = '/images?tag=power&tag=fake'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'tags': ['power', 'fake']})
actual = [image.tags for image in output['images']]
self.assertEqual(0, len(actual))
def test_index_with_tags_and_properties(self):
path = '/images?tag=64bit&hypervisor_type=kvm'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'tags': ['64bit'],
'hypervisor_type': 'kvm'})
tags = [image.tags for image in output['images']]
properties = [image.extra_properties for image in output['images']]
self.assertEqual(len(tags), len(properties))
self.assertIn('64bit', tags[0])
self.assertEqual('kvm', properties[0]['hypervisor_type'])
def test_index_with_multiple_properties(self):
path = '/images?foo=bar&hypervisor_type=kvm'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'foo': 'bar',
'hypervisor_type': 'kvm'})
properties = [image.extra_properties for image in output['images']]
self.assertEqual('kvm', properties[0]['hypervisor_type'])
self.assertEqual('bar', properties[0]['foo'])
def test_index_with_core_and_extra_property(self):
path = '/images?disk_format=raw&foo=bar'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'foo': 'bar',
'disk_format': 'raw'})
properties = [image.extra_properties for image in output['images']]
self.assertEqual(1, len(output['images']))
self.assertEqual('raw', output['images'][0].disk_format)
self.assertEqual('bar', properties[0]['foo'])
def test_index_with_nonexistent_properties(self):
path = '/images?abc=xyz&pudding=banana'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'abc': 'xyz',
'pudding': 'banana'})
self.assertEqual(0, len(output['images']))
def test_index_with_non_existent_tags(self):
path = '/images?tag=fake'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request,
filters={'tags': ['fake']})
actual = [image.tags for image in output['images']]
self.assertEqual(0, len(actual))
def test_show(self):
request = unit_test_utils.get_fake_request()
output = self.controller.show(request, image_id=UUID2)
self.assertEqual(UUID2, output.image_id)
self.assertEqual('2', output.name)
def test_show_deleted_properties(self):
"""Ensure that the api filters out deleted image properties."""
# get the image properties into the odd state
image = {
'id': str(uuid.uuid4()),
'status': 'active',
'properties': {'poo': 'bear'},
}
self.db.image_create(None, image)
self.db.image_update(None, image['id'],
{'properties': {'yin': 'yang'}},
purge_props=True)
request = unit_test_utils.get_fake_request()
output = self.controller.show(request, image['id'])
self.assertEqual('yang', output.extra_properties['yin'])
def test_show_non_existent(self):
request = unit_test_utils.get_fake_request()
image_id = str(uuid.uuid4())
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, request, image_id)
def test_show_deleted_image_admin(self):
request = unit_test_utils.get_fake_request(is_admin=True)
self.controller.delete(request, UUID1)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, request, UUID1)
def test_show_not_allowed(self):
request = unit_test_utils.get_fake_request()
self.assertEqual(TENANT1, request.context.tenant)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, request, UUID4)
def test_create(self):
request = unit_test_utils.get_fake_request()
image = {'name': 'image-1'}
output = self.controller.create(request, image=image,
extra_properties={},
tags=[])
self.assertEqual('image-1', output.name)
self.assertEqual({}, output.extra_properties)
self.assertEqual(set([]), output.tags)
self.assertEqual('private', output.visibility)
output_logs = self.notifier.get_logs()
self.assertEqual(1, len(output_logs))
output_log = output_logs[0]
self.assertEqual('INFO', output_log['notification_type'])
self.assertEqual('image.create', output_log['event_type'])
self.assertEqual('image-1', output_log['payload']['name'])
def test_create_with_properties(self):
request = unit_test_utils.get_fake_request()
image_properties = {'foo': 'bar'}
image = {'name': 'image-1'}
output = self.controller.create(request, image=image,
extra_properties=image_properties,
tags=[])
self.assertEqual('image-1', output.name)
self.assertEqual(image_properties, output.extra_properties)
self.assertEqual(set([]), output.tags)
self.assertEqual('private', output.visibility)
output_logs = self.notifier.get_logs()
self.assertEqual(1, len(output_logs))
output_log = output_logs[0]
self.assertEqual('INFO', output_log['notification_type'])
self.assertEqual('image.create', output_log['event_type'])
self.assertEqual('image-1', output_log['payload']['name'])
def test_create_with_too_many_properties(self):
self.config(image_property_quota=1)
request = unit_test_utils.get_fake_request()
image_properties = {'foo': 'bar', 'foo2': 'bar'}
image = {'name': 'image-1'}
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.create, request,
image=image,
extra_properties=image_properties,
tags=[])
def test_create_with_bad_min_disk_size(self):
request = unit_test_utils.get_fake_request()
image = {'min_disk': -42, 'name': 'image-1'}
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, request,
image=image,
extra_properties={},
tags=[])
def test_create_with_bad_min_ram_size(self):
request = unit_test_utils.get_fake_request()
image = {'min_ram': -42, 'name': 'image-1'}
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, request,
image=image,
extra_properties={},
tags=[])
def test_create_public_image_as_admin(self):
request = unit_test_utils.get_fake_request()
image = {'name': 'image-1', 'visibility': 'public'}
output = self.controller.create(request, image=image,
extra_properties={}, tags=[])
self.assertEqual('public', output.visibility)
output_logs = self.notifier.get_logs()
self.assertEqual(1, len(output_logs))
output_log = output_logs[0]
self.assertEqual('INFO', output_log['notification_type'])
self.assertEqual('image.create', output_log['event_type'])
self.assertEqual(output.image_id, output_log['payload']['id'])
def test_create_dup_id(self):
request = unit_test_utils.get_fake_request()
image = {'image_id': UUID4}
self.assertRaises(webob.exc.HTTPConflict,
self.controller.create,
request,
image=image,
extra_properties={},
tags=[])
def test_create_duplicate_tags(self):
request = unit_test_utils.get_fake_request()
tags = ['ping', 'ping']
output = self.controller.create(request, image={},
extra_properties={}, tags=tags)
self.assertEqual(set(['ping']), output.tags)
output_logs = self.notifier.get_logs()
self.assertEqual(1, len(output_logs))
output_log = output_logs[0]
self.assertEqual('INFO', output_log['notification_type'])
self.assertEqual('image.create', output_log['event_type'])
self.assertEqual(output.image_id, output_log['payload']['id'])
def test_create_with_too_many_tags(self):
self.config(image_tag_quota=1)
request = unit_test_utils.get_fake_request()
tags = ['ping', 'pong']
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.create,
request, image={}, extra_properties={},
tags=tags)
def test_create_with_duplicate_location(self):
request = unit_test_utils.get_fake_request()
location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
image = {'name': 'image-1', 'locations': [location, location]}
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
request, image=image, extra_properties={},
tags=[])
def test_create_unexpected_property(self):
request = unit_test_utils.get_fake_request()
image_properties = {'unexpected': 'unexpected'}
image = {'name': 'image-1'}
with mock.patch.object(domain.ImageFactory, 'new_image',
side_effect=TypeError):
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, request, image=image,
extra_properties=image_properties, tags=[])
def test_create_reserved_property(self):
request = unit_test_utils.get_fake_request()
image_properties = {'reserved': 'reserved'}
image = {'name': 'image-1'}
with mock.patch.object(domain.ImageFactory, 'new_image',
side_effect=exception.ReservedProperty(
property='reserved')):
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, request, image=image,
extra_properties=image_properties, tags=[])
def test_create_readonly_property(self):
request = unit_test_utils.get_fake_request()
image_properties = {'readonly': 'readonly'}
image = {'name': 'image-1'}
with mock.patch.object(domain.ImageFactory, 'new_image',
side_effect=exception.ReadonlyProperty(
property='readonly')):
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create, request, image=image,
extra_properties=image_properties, tags=[])
def test_update_no_changes(self):
request = unit_test_utils.get_fake_request()
output = self.controller.update(request, UUID1, changes=[])
self.assertEqual(UUID1, output.image_id)
self.assertEqual(output.created_at, output.updated_at)
self.assertEqual(2, len(output.tags))
self.assertIn('ping', output.tags)
self.assertIn('pong', output.tags)
output_logs = self.notifier.get_logs()
# NOTE(markwash): don't send a notification if nothing is updated
self.assertEqual(0, len(output_logs))
def test_update_with_bad_min_disk(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['min_disk'], 'value': -42}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes=changes)
def test_update_with_bad_min_ram(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['min_ram'], 'value': -42}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes=changes)
def test_update_image_doesnt_exist(self):
request = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPNotFound, self.controller.update,
request, str(uuid.uuid4()), changes=[])
def test_update_deleted_image_admin(self):
request = unit_test_utils.get_fake_request(is_admin=True)
self.controller.delete(request, UUID1)
self.assertRaises(webob.exc.HTTPNotFound, self.controller.update,
request, UUID1, changes=[])
def test_update_with_too_many_properties(self):
self.config(user_storage_quota='1')
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.update,
request, UUID1, changes=changes)
def test_update_replace_base_attribute(self):
self.db.image_update(None, UUID1, {'properties': {'foo': 'bar'}})
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['name'], 'value': 'fedora'}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual('fedora', output.name)
self.assertEqual({'foo': 'bar'}, output.extra_properties)
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_replace_tags(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'replace', 'path': ['tags'], 'value': ['king', 'kong']},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(output.image_id, UUID1)
self.assertEqual(2, len(output.tags))
self.assertIn('king', output.tags)
self.assertIn('kong', output.tags)
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_replace_property(self):
request = unit_test_utils.get_fake_request()
properties = {'foo': 'bar', 'snitch': 'golden'}
self.db.image_update(None, UUID1, {'properties': properties})
output = self.controller.show(request, UUID1)
self.assertEqual('bar', output.extra_properties['foo'])
self.assertEqual('golden', output.extra_properties['snitch'])
changes = [
{'op': 'replace', 'path': ['foo'], 'value': 'baz'},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual('baz', output.extra_properties['foo'])
self.assertEqual('golden', output.extra_properties['snitch'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_add_too_many_properties(self):
self.config(image_property_quota=1)
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['foo'], 'value': 'baz'},
{'op': 'add', 'path': ['snitch'], 'value': 'golden'},
]
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.update, request,
UUID1, changes)
def test_update_add_and_remove_too_many_properties(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['foo'], 'value': 'baz'},
{'op': 'add', 'path': ['snitch'], 'value': 'golden'},
]
self.controller.update(request, UUID1, changes)
self.config(image_property_quota=1)
# We must remove two properties to avoid being
# over the limit of 1 property
changes = [
{'op': 'remove', 'path': ['foo']},
{'op': 'add', 'path': ['fizz'], 'value': 'buzz'},
]
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.update, request,
UUID1, changes)
def test_update_add_unlimited_properties(self):
self.config(image_property_quota=-1)
request = unit_test_utils.get_fake_request()
output = self.controller.show(request, UUID1)
changes = [{'op': 'add',
'path': ['foo'],
'value': 'bar'}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
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': str(uuid.uuid4()),
'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('raw', resp.disk_format)
self.assertEqual('bare', resp.container_format)
def test_update_remove_property_while_over_limit(self):
"""Ensure that image properties can be removed.
Image properties should be able to be removed as long as the image has
fewer than the limited number of image properties after the
transaction.
"""
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['foo'], 'value': 'baz'},
{'op': 'add', 'path': ['snitch'], 'value': 'golden'},
{'op': 'add', 'path': ['fizz'], 'value': 'buzz'},
]
self.controller.update(request, UUID1, changes)
self.config(image_property_quota=1)
# We must remove two properties to avoid being
# over the limit of 1 property
changes = [
{'op': 'remove', 'path': ['foo']},
{'op': 'remove', 'path': ['snitch']},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.extra_properties))
self.assertEqual('buzz', output.extra_properties['fizz'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_add_and_remove_property_under_limit(self):
"""Ensure that image properties can be removed.
Image properties should be able to be added and removed simultaneously
as long as the image has fewer than the limited number of image
properties after the transaction.
"""
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['foo'], 'value': 'baz'},
{'op': 'add', 'path': ['snitch'], 'value': 'golden'},
]
self.controller.update(request, UUID1, changes)
self.config(image_property_quota=1)
# We must remove two properties to avoid being
# over the limit of 1 property
changes = [
{'op': 'remove', 'path': ['foo']},
{'op': 'remove', 'path': ['snitch']},
{'op': 'add', 'path': ['fizz'], 'value': 'buzz'},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.extra_properties))
self.assertEqual('buzz', output.extra_properties['fizz'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_replace_missing_property(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'replace', 'path': 'foo', 'value': 'baz'},
]
self.assertRaises(webob.exc.HTTPConflict,
self.controller.update, request, UUID1, changes)
def test_prop_protection_with_create_and_permitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
created_image = self.controller.create(request, image=image,
extra_properties={},
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['member'])
changes = [
{'op': 'add', 'path': ['x_owner_foo'], 'value': 'bar'},
]
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertEqual('bar', output.extra_properties['x_owner_foo'])
def test_prop_protection_with_update_and_permitted_policy(self):
self.set_property_protections(use_policies=True)
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
request = unit_test_utils.get_fake_request(roles=['spl_role'])
image = {'name': 'image-1'}
extra_props = {'spl_creator_policy': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
self.assertEqual('bar',
created_image.extra_properties['spl_creator_policy'])
another_request = unit_test_utils.get_fake_request(roles=['spl_role'])
changes = [
{'op': 'replace', 'path': ['spl_creator_policy'], 'value': 'par'},
]
self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
another_request, created_image.image_id, changes)
another_request = unit_test_utils.get_fake_request(roles=['admin'])
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertEqual('par',
output.extra_properties['spl_creator_policy'])
def test_prop_protection_with_create_with_patch_and_policy(self):
self.set_property_protections(use_policies=True)
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
request = unit_test_utils.get_fake_request(roles=['spl_role', 'admin'])
image = {'name': 'image-1'}
extra_props = {'spl_default_policy': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
changes = [
{'op': 'add', 'path': ['spl_creator_policy'], 'value': 'bar'},
]
self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
another_request, created_image.image_id, changes)
another_request = unit_test_utils.get_fake_request(roles=['spl_role'])
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertEqual('bar',
output.extra_properties['spl_creator_policy'])
def test_prop_protection_with_create_and_unpermitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
created_image = self.controller.create(request, image=image,
extra_properties={},
tags=[])
roles = ['fake_member']
another_request = unit_test_utils.get_fake_request(roles=roles)
changes = [
{'op': 'add', 'path': ['x_owner_foo'], 'value': 'bar'},
]
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.update, another_request,
created_image.image_id, changes)
def test_prop_protection_with_show_and_permitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_owner_foo': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['member'])
output = self.controller.show(another_request, created_image.image_id)
self.assertEqual('bar', output.extra_properties['x_owner_foo'])
def test_prop_protection_with_show_and_unpermitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['member'])
image = {'name': 'image-1'}
extra_props = {'x_owner_foo': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
output = self.controller.show(another_request, created_image.image_id)
self.assertRaises(KeyError, output.extra_properties.__getitem__,
'x_owner_foo')
def test_prop_protection_with_update_and_permitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_owner_foo': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['member'])
changes = [
{'op': 'replace', 'path': ['x_owner_foo'], 'value': 'baz'},
]
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertEqual('baz', output.extra_properties['x_owner_foo'])
def test_prop_protection_with_update_and_unpermitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_owner_foo': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
changes = [
{'op': 'replace', 'path': ['x_owner_foo'], 'value': 'baz'},
]
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
another_request, created_image.image_id, changes)
def test_prop_protection_with_delete_and_permitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_owner_foo': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['member'])
changes = [
{'op': 'remove', 'path': ['x_owner_foo']}
]
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertRaises(KeyError, output.extra_properties.__getitem__,
'x_owner_foo')
def test_prop_protection_with_delete_and_unpermitted_role(self):
enforcer = glance.api.policy.Enforcer()
self.controller = glance.api.v2.images.ImagesController(self.db,
enforcer,
self.notifier,
self.store)
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_owner_foo': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
changes = [
{'op': 'remove', 'path': ['x_owner_foo']}
]
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
another_request, created_image.image_id, changes)
def test_create_non_protected_prop(self):
"""Property marked with special char @ creatable by an unknown role"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_all_permitted_1': '1'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
self.assertEqual('1',
created_image.extra_properties['x_all_permitted_1'])
another_request = unit_test_utils.get_fake_request(roles=['joe_soap'])
extra_props = {'x_all_permitted_2': '2'}
created_image = self.controller.create(another_request, image=image,
extra_properties=extra_props,
tags=[])
self.assertEqual('2',
created_image.extra_properties['x_all_permitted_2'])
def test_read_non_protected_prop(self):
"""Property marked with special char @ readable by an unknown role"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_all_permitted': '1'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['joe_soap'])
output = self.controller.show(another_request, created_image.image_id)
self.assertEqual('1', output.extra_properties['x_all_permitted'])
def test_update_non_protected_prop(self):
"""Property marked with special char @ updatable by an unknown role"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_all_permitted': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['joe_soap'])
changes = [
{'op': 'replace', 'path': ['x_all_permitted'], 'value': 'baz'},
]
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertEqual('baz', output.extra_properties['x_all_permitted'])
def test_delete_non_protected_prop(self):
"""Property marked with special char @ deletable by an unknown role"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_all_permitted': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['member'])
changes = [
{'op': 'remove', 'path': ['x_all_permitted']}
]
output = self.controller.update(another_request,
created_image.image_id, changes)
self.assertRaises(KeyError, output.extra_properties.__getitem__,
'x_all_permitted')
def test_create_locked_down_protected_prop(self):
"""Property marked with special char ! creatable by no one"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
created_image = self.controller.create(request, image=image,
extra_properties={},
tags=[])
roles = ['fake_member']
another_request = unit_test_utils.get_fake_request(roles=roles)
changes = [
{'op': 'add', 'path': ['x_none_permitted'], 'value': 'bar'},
]
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.update, another_request,
created_image.image_id, changes)
def test_read_locked_down_protected_prop(self):
"""Property marked with special char ! readable by no one"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['member'])
image = {'name': 'image-1'}
extra_props = {'x_none_read': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
output = self.controller.show(another_request, created_image.image_id)
self.assertRaises(KeyError, output.extra_properties.__getitem__,
'x_none_read')
def test_update_locked_down_protected_prop(self):
"""Property marked with special char ! updatable by no one"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_none_update': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
changes = [
{'op': 'replace', 'path': ['x_none_update'], 'value': 'baz'},
]
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
another_request, created_image.image_id, changes)
def test_delete_locked_down_protected_prop(self):
"""Property marked with special char ! deletable by no one"""
self.set_property_protections()
request = unit_test_utils.get_fake_request(roles=['admin'])
image = {'name': 'image-1'}
extra_props = {'x_none_delete': 'bar'}
created_image = self.controller.create(request, image=image,
extra_properties=extra_props,
tags=[])
another_request = unit_test_utils.get_fake_request(roles=['fake_role'])
changes = [
{'op': 'remove', 'path': ['x_none_delete']}
]
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
another_request, created_image.image_id, changes)
def test_update_replace_locations(self):
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.assertIsNone(output.size)
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location]}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertEqual(new_location, output.locations[0])
self.assertEqual('active', output.status)
def test_update_replace_locations_non_empty(self):
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_replace_locations_invalid(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'],
'value': [{'url': 'unknow://foo', 'metadata': {}}]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_replace_locations_status_exception(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID2, changes)
self.assertEqual(UUID2, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.db.image_update(None, UUID2, {'disk_format': None})
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID2, changes)
def test_update_add_property(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['foo'], 'value': 'baz'},
{'op': 'add', 'path': ['snitch'], 'value': 'golden'},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual('baz', output.extra_properties['foo'])
self.assertEqual('golden', output.extra_properties['snitch'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_add_base_property_json_schema_version_4(self):
request = unit_test_utils.get_fake_request()
changes = [{
'json_schema_version': 4, 'op': 'add',
'path': ['name'], 'value': 'fedora'
}]
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
request, UUID1, changes)
def test_update_add_extra_property_json_schema_version_4(self):
self.db.image_update(None, UUID1, {'properties': {'foo': 'bar'}})
request = unit_test_utils.get_fake_request()
changes = [{
'json_schema_version': 4, 'op': 'add',
'path': ['foo'], 'value': 'baz'
}]
self.assertRaises(webob.exc.HTTPConflict, self.controller.update,
request, UUID1, changes)
def test_update_add_base_property_json_schema_version_10(self):
request = unit_test_utils.get_fake_request()
changes = [{
'json_schema_version': 10, 'op': 'add',
'path': ['name'], 'value': 'fedora'
}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(output.image_id, UUID1)
self.assertEqual(output.name, 'fedora')
def test_update_add_extra_property_json_schema_version_10(self):
self.db.image_update(None, UUID1, {'properties': {'foo': 'bar'}})
request = unit_test_utils.get_fake_request()
changes = [{
'json_schema_version': 10, 'op': 'add',
'path': ['foo'], 'value': 'baz'
}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(output.image_id, UUID1)
self.assertEqual(output.extra_properties, {'foo': 'baz'})
def test_update_add_property_already_present_json_schema_version_4(self):
request = unit_test_utils.get_fake_request()
properties = {'foo': 'bar'}
self.db.image_update(None, UUID1, {'properties': properties})
output = self.controller.show(request, UUID1)
self.assertEqual('bar', output.extra_properties['foo'])
changes = [
{'json_schema_version': 4, 'op': 'add',
'path': ['foo'], 'value': 'baz'},
]
self.assertRaises(webob.exc.HTTPConflict,
self.controller.update, request, UUID1, changes)
def test_update_add_property_already_present_json_schema_version_10(self):
request = unit_test_utils.get_fake_request()
properties = {'foo': 'bar'}
self.db.image_update(None, UUID1, {'properties': properties})
output = self.controller.show(request, UUID1)
self.assertEqual(output.extra_properties['foo'], 'bar')
changes = [
{'json_schema_version': 10, 'op': 'add',
'path': ['foo'], 'value': 'baz'},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(output.image_id, UUID1)
self.assertEqual(output.extra_properties, {'foo': 'baz'})
def test_update_add_locations(self):
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(2, len(output.locations))
self.assertEqual(new_location, output.locations[1])
def test_update_add_locations_insertion(self):
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
changes = [{'op': 'add', 'path': ['locations', '0'],
'value': new_location}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(2, len(output.locations))
self.assertEqual(new_location, output.locations[0])
def test_update_add_locations_list(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': {'url': 'foo', 'metadata': {}}}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_add_locations_invalid(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': {'url': 'unknow://foo', 'metadata': {}}}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
changes = [{'op': 'add', 'path': ['locations', None],
'value': {'url': 'unknow://foo', 'metadata': {}}}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_add_locations_status_exception(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID2, changes)
self.assertEqual(UUID2, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.db.image_update(None, UUID2, {'disk_format': None})
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID2, changes)
def test_update_add_duplicate_locations(self):
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
request = unit_test_utils.get_fake_request()
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(2, len(output.locations))
self.assertEqual(new_location, output.locations[1])
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_replace_duplicate_locations(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'replace', 'path': ['locations'], 'value': []}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'replace', 'path': ['locations'],
'value': [new_location, new_location]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_add_too_many_locations(self):
self.config(image_location_quota=1)
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_1' % BASE_URI,
'metadata': {}}},
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_2' % BASE_URI,
'metadata': {}}},
]
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.update, request,
UUID1, changes)
def test_update_add_and_remove_too_many_locations(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_1' % BASE_URI,
'metadata': {}}},
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_2' % BASE_URI,
'metadata': {}}},
]
self.controller.update(request, UUID1, changes)
self.config(image_location_quota=1)
# We must remove two properties to avoid being
# over the limit of 1 property
changes = [
{'op': 'remove', 'path': ['locations', '0']},
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_3' % BASE_URI,
'metadata': {}}},
]
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.update, request,
UUID1, changes)
def test_update_add_unlimited_locations(self):
self.config(image_location_quota=-1)
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_1' % BASE_URI,
'metadata': {}}},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_remove_location_while_over_limit(self):
"""Ensure that image locations can be removed.
Image locations should be able to be removed as long as the image has
fewer than the limited number of image locations after the
transaction.
"""
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_1' % BASE_URI,
'metadata': {}}},
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_2' % BASE_URI,
'metadata': {}}},
]
self.controller.update(request, UUID1, changes)
self.config(image_location_quota=1)
self.config(show_multiple_locations=True)
# We must remove two locations to avoid being over
# the limit of 1 location
changes = [
{'op': 'remove', 'path': ['locations', '0']},
{'op': 'remove', 'path': ['locations', '0']},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertIn('fake_location_2', output.locations[0]['url'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_add_and_remove_location_under_limit(self):
"""Ensure that image locations can be removed.
Image locations should be able to be added and removed simultaneously
as long as the image has fewer than the limited number of image
locations after the transaction.
"""
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
self.config(show_multiple_locations=True)
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_1' % BASE_URI,
'metadata': {}}},
]
self.controller.update(request, UUID1, changes)
self.config(image_location_quota=1)
# We must remove two properties to avoid being
# over the limit of 1 property
changes = [
{'op': 'remove', 'path': ['locations', '0']},
{'op': 'remove', 'path': ['locations', '0']},
{'op': 'add', 'path': ['locations', '-'],
'value': {'url': '%s/fake_location_3' % BASE_URI,
'metadata': {}}},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertIn('fake_location_3', output.locations[0]['url'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_remove_base_property(self):
self.db.image_update(None, UUID1, {'properties': {'foo': 'bar'}})
request = unit_test_utils.get_fake_request()
changes = [{'op': 'remove', 'path': ['name']}]
self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
request, UUID1, changes)
def test_update_remove_property(self):
request = unit_test_utils.get_fake_request()
properties = {'foo': 'bar', 'snitch': 'golden'}
self.db.image_update(None, UUID1, {'properties': properties})
output = self.controller.show(request, UUID1)
self.assertEqual('bar', output.extra_properties['foo'])
self.assertEqual('golden', output.extra_properties['snitch'])
changes = [
{'op': 'remove', 'path': ['snitch']},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual({'foo': 'bar'}, output.extra_properties)
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_remove_missing_property(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'remove', 'path': ['foo']},
]
self.assertRaises(webob.exc.HTTPConflict,
self.controller.update, request, UUID1, changes)
def test_update_remove_location(self):
self.stubs.Set(store, 'get_size_from_backend',
unit_test_utils.fake_get_size_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'remove', 'path': ['locations', '0']}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(output.image_id, UUID1)
self.assertEqual(0, len(output.locations))
self.assertEqual('queued', output.status)
self.assertIsNone(output.size)
new_location = {'url': '%s/fake_location' % BASE_URI, 'metadata': {}}
changes = [{'op': 'add', 'path': ['locations', '-'],
'value': new_location}]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(1, len(output.locations))
self.assertEqual(new_location, output.locations[0])
self.assertEqual('active', output.status)
def test_update_remove_location_invalid_pos(self):
request = unit_test_utils.get_fake_request()
changes = [{'op': 'remove', 'path': ['locations', None]}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
changes = [{'op': 'remove', 'path': ['locations', '-1']}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
changes = [{'op': 'remove', 'path': ['locations', '99']}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
changes = [{'op': 'remove', 'path': ['locations', 'x']}]
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
request, UUID1, changes)
def test_update_remove_location_store_exception(self):
def fake_delete_image_location_from_backend(self, *args, **kwargs):
raise Exception('fake_backend_exception')
self.stubs.Set(self.store_utils, 'delete_image_location_from_backend',
fake_delete_image_location_from_backend)
request = unit_test_utils.get_fake_request()
changes = [{'op': 'remove', 'path': ['locations', '0']}]
self.assertRaises(webob.exc.HTTPInternalServerError,
self.controller.update, request, UUID1, changes)
def test_update_multiple_changes(self):
request = unit_test_utils.get_fake_request()
properties = {'foo': 'bar', 'snitch': 'golden'}
self.db.image_update(None, UUID1, {'properties': properties})
changes = [
{'op': 'replace', 'path': ['min_ram'], 'value': 128},
{'op': 'replace', 'path': ['foo'], 'value': 'baz'},
{'op': 'remove', 'path': ['snitch']},
{'op': 'add', 'path': ['kb'], 'value': 'dvorak'},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(UUID1, output.image_id)
self.assertEqual(128, output.min_ram)
self.addDetail('extra_properties',
testtools.content.json_content(
jsonutils.dumps(output.extra_properties)))
self.assertEqual(2, len(output.extra_properties))
self.assertEqual('baz', output.extra_properties['foo'])
self.assertEqual('dvorak', output.extra_properties['kb'])
self.assertNotEqual(output.created_at, output.updated_at)
def test_update_invalid_operation(self):
request = unit_test_utils.get_fake_request()
change = {'op': 'test', 'path': 'options', 'value': 'puts'}
try:
self.controller.update(request, UUID1, [change])
except AssertionError:
pass # AssertionError is the desired behavior
else:
self.fail('Failed to raise AssertionError on %s' % change)
def test_update_duplicate_tags(self):
request = unit_test_utils.get_fake_request()
changes = [
{'op': 'replace', 'path': ['tags'], 'value': ['ping', 'ping']},
]
output = self.controller.update(request, UUID1, changes)
self.assertEqual(1, len(output.tags))
self.assertIn('ping', output.tags)
output_logs = self.notifier.get_logs()
self.assertEqual(1, len(output_logs))
output_log = output_logs[0]
self.assertEqual('INFO', output_log['notification_type'])
self.assertEqual('image.update', output_log['event_type'])
self.assertEqual(UUID1, output_log['payload']['id'])
def test_delete(self):
request = unit_test_utils.get_fake_request()
self.assertIn('%s/%s' % (BASE_URI, UUID1), self.store.data)
try:
self.controller.delete(request, UUID1)
output_logs = self.notifier.get_logs()
self.assertEqual(1, len(output_logs))
output_log = output_logs[0]
self.assertEqual('INFO', output_log['notification_type'])
self.assertEqual("image.delete", output_log['event_type'])
except Exception as e:
self.fail("Delete raised exception: %s" % e)
deleted_img = self.db.image_get(request.context, UUID1,
force_show_deleted=True)
self.assertTrue(deleted_img['deleted'])
self.assertEqual('deleted', deleted_img['status'])
self.assertNotIn('%s/%s' % (BASE_URI, UUID1), self.store.data)
def test_delete_queued_updates_status(self):
"""Ensure status of queued image is updated (LP bug #1048851)"""
request = unit_test_utils.get_fake_request(is_admin=True)
image = self.db.image_create(request.context, {'status': 'queued'})
image_id = image['id']
self.controller.delete(request, image_id)
image = self.db.image_get(request.context, image_id,
force_show_deleted=True)
self.assertTrue(image['deleted'])
self.assertEqual('deleted', image['status'])
def test_delete_queued_updates_status_delayed_delete(self):
"""Ensure status of queued image is updated (LP bug #1048851).
Must be set to 'deleted' when delayed_delete isenabled.
"""
scrubber_dir = os.path.join(self.test_dir, 'scrubber')
self.config(delayed_delete=True, scrubber_datadir=scrubber_dir)
request = unit_test_utils.get_fake_request(is_admin=True)
image = self.db.image_create(request.context, {'status': 'queued'})
image_id = image['id']
self.controller.delete(request, image_id)
image = self.db.image_get(request.context, image_id,
force_show_deleted=True)
self.assertTrue(image['deleted'])