# -*- coding: utf-8 -*-
# Copyright 2010-2011 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 uuid
import mock
from oslo_serialization import jsonutils
import routes
import six
from six.moves import http_client as http
import webob
import glance.api.common
import glance.common.config
from glance.common import crypt
from glance.common import timeutils
from glance import context
from glance.db.sqlalchemy import api as db_api
from glance.db.sqlalchemy import models as db_models
from glance.registry.api import v1 as rserver
from glance.tests.unit import base
from glance.tests import utils as test_utils
_gen_uuid = lambda: str(uuid.uuid4())
UUID1 = _gen_uuid()
UUID2 = _gen_uuid()
[docs]class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
[docs] def setUp(self):
"""Establish a clean test environment"""
super(TestRegistryAPI, self).setUp()
self.mapper = routes.Mapper()
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=True)
def _get_extra_fixture(id, name, **kwargs):
return self.get_extra_fixture(
id, name,
locations=[{'url': "file:///%s/%s" % (self.test_dir, id),
'metadata': {}, 'status': 'active'}], **kwargs)
self.FIXTURES = [
_get_extra_fixture(UUID1, 'fake image #1', is_public=False,
disk_format='ami', container_format='ami',
min_disk=0, min_ram=0, owner=123,
size=13, properties={'type': 'kernel'}),
_get_extra_fixture(UUID2, 'fake image #2',
min_disk=5, min_ram=256,
size=19, properties={})]
self.context = context.RequestContext(is_admin=True)
db_api.get_engine()
self.destroy_fixtures()
self.create_fixtures()
[docs] def tearDown(self):
"""Clear the test environment"""
super(TestRegistryAPI, self).tearDown()
self.destroy_fixtures()
[docs] def test_show(self):
"""
Tests that the /images/<id> registry API endpoint
returns the expected image
"""
fixture = {'id': UUID2,
'name': 'fake image #2',
'size': 19,
'min_ram': 256,
'min_disk': 5,
'checksum': None}
res = self.get_api_response_ext(http.OK, '/images/%s' % UUID2)
res_dict = jsonutils.loads(res.body)
image = res_dict['image']
for k, v in six.iteritems(fixture):
self.assertEqual(v, image[k])
[docs] def test_show_unknown(self):
"""
Tests that the /images/<id> registry API endpoint
returns a 404 for an unknown image id
"""
self.get_api_response_ext(http.NOT_FOUND, '/images/%s' % _gen_uuid())
[docs] def test_show_invalid(self):
"""
Tests that the /images/<id> registry API endpoint
returns a 404 for an invalid (therefore unknown) image id
"""
self.get_api_response_ext(http.NOT_FOUND, '/images/%s' % _gen_uuid())
[docs] def test_show_deleted_image_as_admin(self):
"""
Tests that the /images/<id> registry API endpoint
returns a 200 for deleted image to admin user.
"""
# Delete image #2
self.get_api_response_ext(http.OK, '/images/%s' % UUID2,
method='DELETE')
self.get_api_response_ext(http.OK, '/images/%s' % UUID2)
[docs] def test_show_deleted_image_as_nonadmin(self):
"""
Tests that the /images/<id> registry API endpoint
returns a 404 for deleted image to non-admin user.
"""
# Delete image #2
self.get_api_response_ext(http.OK, '/images/%s' % UUID2,
method='DELETE')
api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=False)
self.get_api_response_ext(http.NOT_FOUND, '/images/%s' % UUID2,
api=api)
[docs] def test_show_private_image_with_no_admin_user(self):
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, size=18, owner='test user',
is_public=False)
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
self.get_api_response_ext(http.NOT_FOUND, '/images/%s' % UUID4,
api=api)
[docs] def test_get_root(self):
"""
Tests that the root registry API returns "index",
which is a list of public images
"""
fixture = {'id': UUID2, 'size': 19, 'checksum': None}
res = self.get_api_response_ext(http.OK, url='/')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for k, v in six.iteritems(fixture):
self.assertEqual(v, images[0][k])
[docs] def test_get_index(self):
"""
Tests that the /images registry API returns list of
public images
"""
fixture = {'id': UUID2, 'size': 19, 'checksum': None}
res = self.get_api_response_ext(http.OK)
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for k, v in six.iteritems(fixture):
self.assertEqual(v, images[0][k])
[docs] def test_get_index_marker(self):
"""
Tests that the /images registry API returns list of
public images that conforms to a marker query param
"""
time1 = timeutils.utcnow() + datetime.timedelta(seconds=5)
time2 = timeutils.utcnow() + datetime.timedelta(seconds=4)
time3 = timeutils.utcnow()
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=19, created_at=time1)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, created_at=time2)
db_api.image_create(self.context, extra_fixture)
UUID5 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID5, created_at=time3)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images?marker=%s' % UUID4)
self.assertEqualImages(res, (UUID5, UUID2))
[docs] def test_get_index_unknown_marker(self):
"""
Tests that the /images registry API returns a 400
when an unknown marker is provided
"""
self.get_api_response_ext(http.BAD_REQUEST,
url='/images?marker=%s' % _gen_uuid())
[docs] def test_get_index_forbidden_marker(self):
"""
Tests that the /images registry API returns a 400
when a forbidden marker is provided
"""
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
self.get_api_response_ext(http.BAD_REQUEST,
url='/images?marker=%s' % UUID1, api=api)
[docs] def test_get_index_limit(self):
"""
Tests that the /images registry API returns list of
public images that conforms to a limit query param
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=19)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url='/images?limit=1')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
# expect list to be sorted by created_at desc
self.assertEqual(UUID4, images[0]['id'])
[docs] def test_get_index_limit_negative(self):
"""
Tests that the /images registry API returns list of
public images that conforms to a limit query param
"""
self.get_api_response_ext(http.BAD_REQUEST, url='/images?limit=-1')
[docs] def test_get_index_limit_non_int(self):
"""
Tests that the /images registry API returns list of
public images that conforms to a limit query param
"""
self.get_api_response_ext(http.BAD_REQUEST, url='/images?limit=a')
[docs] def test_get_index_limit_marker(self):
"""
Tests that the /images registry API returns list of
public images that conforms to limit and marker query params
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=19)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid())
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(
http.OK, url='/images?marker=%s&limit=1' % UUID3)
self.assertEqualImages(res, (UUID2,))
[docs] def test_get_index_filter_on_user_defined_properties(self):
"""
Tests that /images registry API returns list of public images based
a filter on user-defined properties.
"""
image1_id = _gen_uuid()
properties = {'distro': 'ubuntu', 'arch': 'i386'}
extra_fixture = self.get_fixture(id=image1_id, name='image-extra-1',
properties=properties)
db_api.image_create(self.context, extra_fixture)
image2_id = _gen_uuid()
properties = {'distro': 'ubuntu', 'arch': 'x86_64', 'foo': 'bar'}
extra_fixture = self.get_fixture(id=image2_id, name='image-extra-2',
properties=properties)
db_api.image_create(self.context, extra_fixture)
# Test index with filter containing one user-defined property.
# Filter is 'property-distro=ubuntu'.
# Verify both image1 and image2 are returned
res = self.get_api_response_ext(http.OK, url='/images?'
'property-distro=ubuntu')
images = jsonutils.loads(res.body)['images']
self.assertEqual(2, len(images))
self.assertEqual(image2_id, images[0]['id'])
self.assertEqual(image1_id, images[1]['id'])
# Test index with filter containing one user-defined property but
# non-existent value. Filter is 'property-distro=fedora'.
# Verify neither images are returned
res = self.get_api_response_ext(http.OK, url='/images?'
'property-distro=fedora')
images = jsonutils.loads(res.body)['images']
self.assertEqual(0, len(images))
# Test index with filter containing one user-defined property but
# unique value. Filter is 'property-arch=i386'.
# Verify only image1 is returned.
res = self.get_api_response_ext(http.OK, url='/images?'
'property-arch=i386')
images = jsonutils.loads(res.body)['images']
self.assertEqual(1, len(images))
self.assertEqual(image1_id, images[0]['id'])
# Test index with filter containing one user-defined property but
# unique value. Filter is 'property-arch=x86_64'.
# Verify only image1 is returned.
res = self.get_api_response_ext(http.OK, url='/images?'
'property-arch=x86_64')
images = jsonutils.loads(res.body)['images']
self.assertEqual(1, len(images))
self.assertEqual(image2_id, images[0]['id'])
# Test index with filter containing unique user-defined property.
# Filter is 'property-foo=bar'.
# Verify only image2 is returned.
res = self.get_api_response_ext(http.OK,
url='/images?property-foo=bar')
images = jsonutils.loads(res.body)['images']
self.assertEqual(1, len(images))
self.assertEqual(image2_id, images[0]['id'])
# Test index with filter containing unique user-defined property but
# .value is non-existent. Filter is 'property-foo=baz'.
# Verify neither images are returned.
res = self.get_api_response_ext(http.OK,
url='/images?property-foo=baz')
images = jsonutils.loads(res.body)['images']
self.assertEqual(0, len(images))
# Test index with filter containing multiple user-defined properties
# Filter is 'property-arch=x86_64&property-distro=ubuntu'.
# Verify only image2 is returned.
res = self.get_api_response_ext(http.OK, url='/images?'
'property-arch=x86_64&'
'property-distro=ubuntu')
images = jsonutils.loads(res.body)['images']
self.assertEqual(1, len(images))
self.assertEqual(image2_id, images[0]['id'])
# Test index with filter containing multiple user-defined properties
# Filter is 'property-arch=i386&property-distro=ubuntu'.
# Verify only image1 is returned.
res = self.get_api_response_ext(http.OK,
url='/images?property-arch=i386&'
'property-distro=ubuntu')
images = jsonutils.loads(res.body)['images']
self.assertEqual(1, len(images))
self.assertEqual(image1_id, images[0]['id'])
# Test index with filter containing multiple user-defined properties.
# Filter is 'property-arch=random&property-distro=ubuntu'.
# Verify neither images are returned.
res = self.get_api_response_ext(http.OK, url='/images?'
'property-arch=random&'
'property-distro=ubuntu')
images = jsonutils.loads(res.body)['images']
self.assertEqual(0, len(images))
# Test index with filter containing multiple user-defined properties.
# Filter is 'property-arch=random&property-distro=random'.
# Verify neither images are returned.
res = self.get_api_response_ext(http.OK, url='/images?'
'property-arch=random&'
'property-distro=random')
images = jsonutils.loads(res.body)['images']
self.assertEqual(0, len(images))
# Test index with filter containing multiple user-defined properties.
# Filter is 'property-boo=far&property-poo=far'.
# Verify neither images are returned.
res = self.get_api_response_ext(http.OK,
url='/images?property-boo=far&'
'property-poo=far')
images = jsonutils.loads(res.body)['images']
self.assertEqual(0, len(images))
# Test index with filter containing multiple user-defined properties.
# Filter is 'property-foo=bar&property-poo=far'.
# Verify neither images are returned.
res = self.get_api_response_ext(http.OK,
url='/images?property-foo=bar&'
'property-poo=far')
images = jsonutils.loads(res.body)['images']
self.assertEqual(0, len(images))
[docs] def test_get_index_filter_name(self):
"""
Tests that the /images registry API returns list of
public images that have a specific name. This is really a sanity
check, filtering is tested more in-depth using /images/detail
"""
extra_fixture = self.get_fixture(id=_gen_uuid(),
name='new name! #123', size=19)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), name='new name! #123')
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images?name=new name! #123')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(2, len(images))
for image in images:
self.assertEqual('new name! #123', image['name'])
[docs] def test_get_index_sort_default_created_at_desc(self):
"""
Tests that the /images registry API returns list of
public images that conforms to a default sort key/dir
"""
time1 = timeutils.utcnow() + datetime.timedelta(seconds=5)
time2 = timeutils.utcnow() + datetime.timedelta(seconds=4)
time3 = timeutils.utcnow()
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=19, created_at=time1)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, created_at=time2)
db_api.image_create(self.context, extra_fixture)
UUID5 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID5, created_at=time3)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url='/images')
self.assertEqualImages(res, (UUID3, UUID4, UUID5, UUID2))
[docs] def test_get_index_bad_sort_key(self):
"""Ensure a 400 is returned when a bad sort_key is provided."""
self.get_api_response_ext(http.BAD_REQUEST,
url='/images?sort_key=asdf')
[docs] def test_get_index_bad_sort_dir(self):
"""Ensure a 400 is returned when a bad sort_dir is provided."""
self.get_api_response_ext(http.BAD_REQUEST,
url='/images?sort_dir=asdf')
[docs] def test_get_index_null_name(self):
"""Check 200 is returned when sort_key is null name
Check 200 is returned when sort_key is name and name is null
for specified marker
"""
UUID6 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID6, name=None)
db_api.image_create(self.context, extra_fixture)
self.get_api_response_ext(
http.OK, url='/images?sort_key=name&marker=%s' % UUID6)
[docs] def test_get_index_sort_name_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted alphabetically by name in
ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf', size=19)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz')
db_api.image_create(self.context, extra_fixture)
url = '/images?sort_key=name&sort_dir=asc'
res = self.get_api_response_ext(http.OK, url=url)
self.assertEqualImages(res, (UUID3, UUID2, UUID4))
[docs] def test_get_index_sort_status_desc(self):
"""
Tests that the /images registry API returns list of
public images sorted alphabetically by status in
descending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, status='queued', size=19)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url=(
'/images?sort_key=status&sort_dir=desc'))
self.assertEqualImages(res, (UUID3, UUID4, UUID2))
[docs] def test_get_index_sort_size_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted by size in ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, disk_format='ami',
container_format='ami', size=100)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, disk_format='iso',
container_format='bare', size=2)
db_api.image_create(self.context, extra_fixture)
url = '/images?sort_key=size&sort_dir=asc'
res = self.get_api_response_ext(http.OK, url=url)
self.assertEqualImages(res, (UUID4, UUID2, UUID3))
[docs] def test_get_index_sort_created_at_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted by created_at in ascending order.
"""
now = timeutils.utcnow()
time1 = now + datetime.timedelta(seconds=5)
time2 = now
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, created_at=time1, size=19)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, created_at=time2)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url=(
'/images?sort_key=created_at&sort_dir=asc'))
self.assertEqualImages(res, (UUID2, UUID4, UUID3))
[docs] def test_get_index_sort_updated_at_desc(self):
"""
Tests that the /images registry API returns list of
public images sorted by updated_at in descending order.
"""
now = timeutils.utcnow()
time1 = now + datetime.timedelta(seconds=5)
time2 = now
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=19, created_at=None,
updated_at=time1)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, created_at=None,
updated_at=time2)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url=(
'/images?sort_key=updated_at&sort_dir=desc'))
self.assertEqualImages(res, (UUID3, UUID4, UUID2))
[docs] def test_get_details(self):
"""
Tests that the /images/detail registry API returns
a mapping containing a list of detailed image information
"""
fixture = {'id': UUID2,
'name': 'fake image #2',
'is_public': True,
'size': 19,
'min_disk': 5,
'min_ram': 256,
'checksum': None,
'disk_format': 'vhd',
'container_format': 'ovf',
'status': 'active'}
res = self.get_api_response_ext(http.OK, url='/images/detail')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for k, v in six.iteritems(fixture):
self.assertEqual(v, images[0][k])
[docs] def test_get_details_limit_marker(self):
"""
Tests that the /images/details registry API returns list of
public images that conforms to limit and marker query params.
This functionality is tested more thoroughly on /images, this is
just a sanity check
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=20)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid())
db_api.image_create(self.context, extra_fixture)
url = '/images/detail?marker=%s&limit=1' % UUID3
res = self.get_api_response_ext(http.OK, url=url)
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
# expect list to be sorted by created_at desc
self.assertEqual(UUID2, images[0]['id'])
[docs] def test_get_details_invalid_marker(self):
"""
Tests that the /images/detail registry API returns a 400
when an invalid marker is provided
"""
url = '/images/detail?marker=%s' % _gen_uuid()
self.get_api_response_ext(http.BAD_REQUEST, url=url)
[docs] def test_get_details_forbidden_marker(self):
"""
Tests that the /images/detail registry API returns a 400
when a forbidden marker is provided
"""
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
self.get_api_response_ext(http.BAD_REQUEST, api=api,
url='/images/detail?marker=%s' % UUID1)
[docs] def test_get_details_filter_name(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a specific name
"""
extra_fixture = self.get_fixture(id=_gen_uuid(),
name='new name! #123', size=20)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(),
name='new name! #123')
db_api.image_create(self.context, extra_fixture)
url = '/images/detail?name=new name! #123'
res = self.get_api_response_ext(http.OK, url=url)
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(2, len(images))
for image in images:
self.assertEqual('new name! #123', image['name'])
[docs] def test_get_details_filter_status(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a specific status
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), size=19,
status='active')
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?status=saving')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('saving', image['status'])
[docs] def test_get_details_filter_min_disk(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a specific min_disk
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), min_disk=7, size=19)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), disk_format='ami',
container_format='ami', size=19)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?min_disk=7')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for image in images:
self.assertEqual(7, image['min_disk'])
[docs] def test_get_details_filter_min_ram(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a specific min_ram
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), min_ram=514, size=19)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), disk_format='ami',
container_format='ami', size=19)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?min_ram=514')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for image in images:
self.assertEqual(514, image['min_ram'])
[docs] def test_get_details_filter_size_min(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a size greater than or equal to size_min
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), size=18)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), disk_format='ami',
container_format='ami')
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?size_min=19')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(2, len(images))
for image in images:
self.assertGreaterEqual(image['size'], 19)
[docs] def test_get_details_filter_size_max(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a size less than or equal to size_max
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), size=18)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), disk_format='ami',
container_format='ami')
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?size_max=19')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(2, len(images))
for image in images:
self.assertLessEqual(image['size'], 19)
[docs] def test_get_details_filter_size_min_max(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a size less than or equal to size_max
and greater than or equal to size_min
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), size=18)
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), disk_format='ami',
container_format='ami')
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), size=6)
db_api.image_create(self.context, extra_fixture)
url = '/images/detail?size_min=18&size_max=19'
res = self.get_api_response_ext(http.OK, url=url)
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(2, len(images))
for image in images:
self.assertTrue(18 <= image['size'] <= 19)
[docs] def test_get_details_filter_changes_since(self):
"""
Tests that the /images/detail registry API returns list of
images that changed since the time defined by changes-since
"""
dt1 = timeutils.utcnow() - datetime.timedelta(1)
iso1 = timeutils.isotime(dt1)
date_only1 = dt1.strftime('%Y-%m-%d')
date_only2 = dt1.strftime('%Y%m%d')
date_only3 = dt1.strftime('%Y-%m%d')
dt2 = timeutils.utcnow() + datetime.timedelta(1)
iso2 = timeutils.isotime(dt2)
image_ts = timeutils.utcnow() + datetime.timedelta(2)
hour_before = image_ts.strftime('%Y-%m-%dT%H:%M:%S%%2B01:00')
hour_after = image_ts.strftime('%Y-%m-%dT%H:%M:%S-01:00')
dt4 = timeutils.utcnow() + datetime.timedelta(3)
iso4 = timeutils.isotime(dt4)
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, size=18)
db_api.image_create(self.context, extra_fixture)
db_api.image_destroy(self.context, UUID3)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4,
disk_format='ami',
container_format='ami',
created_at=image_ts,
updated_at=image_ts)
db_api.image_create(self.context, extra_fixture)
# Check a standard list, 4 images in db (2 deleted)
res = self.get_api_response_ext(http.OK, url='/images/detail')
self.assertEqualImages(res, (UUID4, UUID2))
# Expect 3 images (1 deleted)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?changes-since=%s' % iso1))
self.assertEqualImages(res, (UUID4, UUID3, UUID2))
# Expect 1 images (0 deleted)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?changes-since=%s' % iso2))
self.assertEqualImages(res, (UUID4,))
# Expect 1 images (0 deleted)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?changes-since=%s' % hour_before))
self.assertEqualImages(res, (UUID4,))
# Expect 0 images (0 deleted)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?changes-since=%s' % hour_after))
self.assertEqualImages(res, ())
# Expect 0 images (0 deleted)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?changes-since=%s' % iso4))
self.assertEqualImages(res, ())
for param in [date_only1, date_only2, date_only3]:
# Expect 3 images (1 deleted)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?changes-since=%s' % param))
self.assertEqualImages(res, (UUID4, UUID3, UUID2))
# Bad request (empty changes-since param)
self.get_api_response_ext(http.BAD_REQUEST,
url='/images/detail?changes-since=')
[docs] def test_get_details_filter_property(self):
"""
Tests that the /images/detail registry API returns list of
public images that have a specific custom property
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), size=19,
properties={'prop_123': 'v a'})
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), size=19,
disk_format='ami',
container_format='ami',
properties={'prop_123': 'v b'})
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?property-prop_123=v%20a'))
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('v a', image['properties']['prop_123'])
[docs] def test_get_details_filter_public_none(self):
"""
Tests that the /images/detail registry API returns list of
all images if is_public none is passed
"""
extra_fixture = self.get_fixture(id=_gen_uuid(),
is_public=False, size=18)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?is_public=None')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(3, len(images))
[docs] def test_get_details_filter_public_false(self):
"""
Tests that the /images/detail registry API returns list of
private images if is_public false is passed
"""
extra_fixture = self.get_fixture(id=_gen_uuid(),
is_public=False, size=18)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?is_public=False')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(2, len(images))
for image in images:
self.assertEqual(False, image['is_public'])
[docs] def test_get_details_filter_public_true(self):
"""
Tests that the /images/detail registry API returns list of
public images if is_public true is passed (same as default)
"""
extra_fixture = self.get_fixture(id=_gen_uuid(),
is_public=False, size=18)
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?is_public=True')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
for image in images:
self.assertTrue(image['is_public'])
[docs] def test_get_details_filter_deleted_false(self):
"""
Test that the /images/detail registry
API return list of images with deleted filter = false
"""
extra_fixture = {'id': _gen_uuid(),
'status': 'active',
'disk_format': 'vhd',
'container_format': 'ovf',
'name': 'test deleted filter 1',
'size': 18,
'deleted': False,
'checksum': None}
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK,
url='/images/detail?deleted=False')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
for image in images:
self.assertFalse(image['deleted'])
[docs] def test_get_filter_no_public_with_no_admin(self):
"""
Tests that the /images/detail registry API returns list of
public images if is_public true is passed (same as default)
"""
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4,
is_public=False, size=18)
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
res = self.get_api_response_ext(http.OK, api=api,
url='/images/detail?is_public=False')
res_dict = jsonutils.loads(res.body)
images = res_dict['images']
self.assertEqual(1, len(images))
# Check that for non admin user only is_public = True images returns
for image in images:
self.assertTrue(image['is_public'])
[docs] def test_get_filter_protected_with_None_value(self):
"""
Tests that the /images/detail registry API returns 400 error
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), size=18,
protected="False")
db_api.image_create(self.context, extra_fixture)
self.get_api_response_ext(http.BAD_REQUEST,
url='/images/detail?protected=')
[docs] def test_get_filter_protected_with_True_value(self):
"""
Tests that the /images/detail registry API returns 400 error
"""
extra_fixture = self.get_fixture(id=_gen_uuid(),
size=18, protected="True")
db_api.image_create(self.context, extra_fixture)
self.get_api_response_ext(http.OK, url='/images/detail?protected=True')
[docs] def test_get_details_sort_name_asc(self):
"""
Tests that the /images/details registry API returns list of
public images sorted alphabetically by name in
ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf', size=19)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz')
db_api.image_create(self.context, extra_fixture)
res = self.get_api_response_ext(http.OK, url=(
'/images/detail?sort_key=name&sort_dir=asc'))
self.assertEqualImages(res, (UUID3, UUID2, UUID4))
[docs] def test_create_image(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture(is_public=True)
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.OK, body=body,
method='POST', content_type='json')
res_dict = jsonutils.loads(res.body)
for k, v in six.iteritems(fixture):
self.assertEqual(v, res_dict['image'][k])
# Test status was updated properly
self.assertEqual('active', res_dict['image']['status'])
[docs] def test_create_image_with_min_disk(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture(min_disk=5)
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.OK, body=body,
method='POST', content_type='json')
res_dict = jsonutils.loads(res.body)
self.assertEqual(5, res_dict['image']['min_disk'])
[docs] def test_create_image_with_min_ram(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture(min_ram=256)
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.OK, body=body,
method='POST', content_type='json')
res_dict = jsonutils.loads(res.body)
self.assertEqual(256, res_dict['image']['min_ram'])
[docs] def test_create_image_with_min_ram_default(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture()
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.OK, body=body,
method='POST', content_type='json')
res_dict = jsonutils.loads(res.body)
self.assertEqual(0, res_dict['image']['min_ram'])
[docs] def test_create_image_with_min_disk_default(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture()
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.OK, body=body,
method='POST', content_type='json')
res_dict = jsonutils.loads(res.body)
self.assertEqual(0, res_dict['image']['min_disk'])
[docs] def test_create_image_with_bad_status(self):
"""Tests proper exception is raised if a bad status is set"""
fixture = self.get_minimal_fixture(id=_gen_uuid(), status='bad status')
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.BAD_REQUEST, body=body,
method='POST', content_type='json')
self.assertIn(b'Invalid image status', res.body)
[docs] def test_create_image_with_bad_id(self):
"""Tests proper exception is raised if a bad disk_format is set"""
fixture = self.get_minimal_fixture(id='asdf')
body = jsonutils.dump_as_bytes(dict(image=fixture))
self.get_api_response_ext(http.BAD_REQUEST, content_type='json',
method='POST', body=body)
[docs] def test_create_image_with_image_id_in_log(self):
"""Tests correct image id in log message when creating image"""
fixture = self.get_minimal_fixture(
id='0564c64c-3545-4e34-abfb-9d18e5f2f2f9')
self.log_image_id = False
def fake_log_info(msg, image_data):
if ('0564c64c-3545-4e34-abfb-9d18e5f2f2f9' == image_data['id'] and
'Successfully created image' in msg):
self.log_image_id = True
self.stubs.Set(rserver.images.LOG, 'info', fake_log_info)
body = jsonutils.dump_as_bytes(dict(image=fixture))
self.get_api_response_ext(http.OK, content_type='json', method='POST',
body=body)
self.assertTrue(self.log_image_id)
[docs] def test_update_image(self):
"""Tests that the /images PUT registry API updates the image"""
fixture = {'name': 'fake public image #2',
'min_disk': 5,
'min_ram': 256,
'disk_format': 'raw'}
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.OK, url='/images/%s' % UUID2,
body=body, method='PUT',
content_type='json')
res_dict = jsonutils.loads(res.body)
self.assertNotEqual(res_dict['image']['created_at'],
res_dict['image']['updated_at'])
for k, v in six.iteritems(fixture):
self.assertEqual(v, res_dict['image'][k])
@mock.patch.object(rserver.images.LOG, 'debug')
[docs] def test_update_image_not_log_sensitive_info(self, log_debug):
"""
Tests that there is no any sensitive info of image location
was logged in glance during the image update operation.
"""
def fake_log_debug(fmt_str, image_meta):
self.assertNotIn("'locations'", fmt_str % image_meta)
fixture = {'name': 'fake public image #2',
'min_disk': 5,
'min_ram': 256,
'disk_format': 'raw',
'location': 'fake://image'}
body = jsonutils.dump_as_bytes(dict(image=fixture))
log_debug.side_effect = fake_log_debug
res = self.get_api_response_ext(http.OK, url='/images/%s' % UUID2,
body=body, method='PUT',
content_type='json')
res_dict = jsonutils.loads(res.body)
self.assertNotEqual(res_dict['image']['created_at'],
res_dict['image']['updated_at'])
for k, v in six.iteritems(fixture):
self.assertEqual(v, res_dict['image'][k])
[docs] def test_update_image_not_existing(self):
"""
Tests proper exception is raised if attempt to update
non-existing image
"""
fixture = {'status': 'killed'}
body = jsonutils.dump_as_bytes(dict(image=fixture))
self.get_api_response_ext(http.NOT_FOUND,
url='/images/%s' % _gen_uuid(),
method='PUT', body=body, content_type='json')
[docs] def test_update_image_with_bad_status(self):
"""Tests that exception raised trying to set a bad status"""
fixture = {'status': 'invalid'}
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(http.BAD_REQUEST, method='PUT',
body=body,
url='/images/%s' % UUID2,
content_type='json')
self.assertIn(b'Invalid image status', res.body)
[docs] def test_update_private_image_no_admin(self):
"""
Tests proper exception is raised if attempt to update
private image with non admin user, that not belongs to it
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, is_public=False,
protected=True, owner='test user')
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
body = jsonutils.dump_as_bytes(dict(image=extra_fixture))
self.get_api_response_ext(http.NOT_FOUND, body=body, api=api,
url='/images/%s' % UUID8, method='PUT',
content_type='json')
[docs] def test_delete_image(self):
"""Tests that the /images DELETE registry API deletes the image"""
# Grab the original number of images
res = self.get_api_response_ext(http.OK)
res_dict = jsonutils.loads(res.body)
orig_num_images = len(res_dict['images'])
# Delete image #2
self.get_api_response_ext(http.OK, url='/images/%s' % UUID2,
method='DELETE')
# Verify one less image
res = self.get_api_response_ext(http.OK)
res_dict = jsonutils.loads(res.body)
new_num_images = len(res_dict['images'])
self.assertEqual(orig_num_images - 1, new_num_images)
[docs] def test_delete_image_response(self):
"""Tests that the registry API delete returns the image metadata"""
image = self.FIXTURES[0]
res = self.get_api_response_ext(http.OK,
url='/images/%s' % image['id'],
method='DELETE')
deleted_image = jsonutils.loads(res.body)['image']
self.assertEqual(image['id'], deleted_image['id'])
self.assertTrue(deleted_image['deleted'])
self.assertTrue(deleted_image['deleted_at'])
[docs] def test_delete_image_not_existing(self):
"""
Tests proper exception is raised if attempt to delete
non-existing image
"""
self.get_api_response_ext(http.NOT_FOUND,
url='/images/%s' % _gen_uuid(),
method='DELETE')
[docs] def test_delete_public_image_no_admin(self):
"""
Tests proper exception is raised if attempt to delete
public image with non admin user
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=True,
owner='test user')
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
self.get_api_response_ext(http.FORBIDDEN, url='/images/%s' % UUID8,
method='DELETE', api=api)
[docs] def test_delete_private_image_no_admin(self):
"""
Tests proper exception is raised if attempt to delete
private image with non admin user, that not belongs to it
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, is_public=False, size=19,
protected=True, owner='test user')
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
self.get_api_response_ext(http.NOT_FOUND, url='/images/%s' % UUID8,
method='DELETE', api=api)
[docs] def test_get_image_members(self):
"""
Tests members listing for existing images
"""
res = self.get_api_response_ext(http.OK,
url='/images/%s/members' % UUID2,
method='GET')
memb_list = jsonutils.loads(res.body)
num_members = len(memb_list['members'])
self.assertEqual(0, num_members)
[docs] def test_get_image_members_not_existing(self):
"""
Tests proper exception is raised if attempt to get members of
non-existing image
"""
self.get_api_response_ext(http.NOT_FOUND, method='GET',
url='/images/%s/members' % _gen_uuid())
[docs] def test_get_image_members_forbidden(self):
"""
Tests proper exception is raised if attempt to get members of
non-existing image
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, is_public=False, size=19,
protected=True, owner='test user')
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
self.get_api_response_ext(http.NOT_FOUND,
url='/images/%s/members' % UUID8,
method='GET', api=api)
[docs] def test_get_member_images(self):
"""
Tests image listing for members
"""
res = self.get_api_response_ext(http.OK,
url='/shared-images/pattieblack',
method='GET')
memb_list = jsonutils.loads(res.body)
num_members = len(memb_list['shared_images'])
self.assertEqual(0, num_members)
[docs] def test_replace_members(self):
"""
Tests replacing image members raises right exception
"""
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=False)
fixture = dict(member_id='pattieblack')
body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
self.get_api_response_ext(http.UNAUTHORIZED, method='PUT', body=body,
url='/images/%s/members' % UUID2,
content_type='json')
[docs] def test_update_all_image_members_non_existing_image_id(self):
"""
Test update image members raises right exception
"""
# Update all image members
fixture = dict(member_id='test1')
req = webob.Request.blank('/images/%s/members' % _gen_uuid())
req.method = 'PUT'
self.context.tenant = 'test2'
req.content_type = 'application/json'
req.body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
res = req.get_response(self.api)
self.assertEqual(http.NOT_FOUND, res.status_int)
[docs] def test_update_all_image_members_invalid_membership_association(self):
"""
Test update image members raises right exception
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
# Add several members to image
req = webob.Request.blank('/images/%s/members/test1' % UUID8)
req.method = 'PUT'
res = req.get_response(self.api)
# Get all image members:
res = self.get_api_response_ext(http.OK,
url='/images/%s/members' % UUID8,
method='GET')
memb_list = jsonutils.loads(res.body)
num_members = len(memb_list['members'])
self.assertEqual(1, num_members)
fixture = dict(member_id='test1')
body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
self.get_api_response_ext(http.BAD_REQUEST,
url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
[docs] def test_update_all_image_members_non_shared_image_forbidden(self):
"""
Test update image members raises right exception
"""
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
UUID9 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID9, size=19, protected=False)
db_api.image_create(self.context, extra_fixture)
fixture = dict(member_id='test1')
req = webob.Request.blank('/images/%s/members' % UUID9)
req.headers['X-Auth-Token'] = 'test1:test1:'
req.method = 'PUT'
req.content_type = 'application/json'
req.body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
res = req.get_response(api)
self.assertEqual(http.FORBIDDEN, res.status_int)
[docs] def test_update_all_image_members(self):
"""
Test update non existing image members
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
# Add several members to image
req = webob.Request.blank('/images/%s/members/test1' % UUID8)
req.method = 'PUT'
req.get_response(self.api)
fixture = [dict(member_id='test2', can_share=True)]
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(http.NO_CONTENT,
url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
[docs] def test_update_all_image_members_bad_request(self):
"""
Test that right exception is raises
in case if wrong memberships association is supplied
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
# Add several members to image
req = webob.Request.blank('/images/%s/members/test1' % UUID8)
req.method = 'PUT'
req.get_response(self.api)
fixture = dict(member_id='test3')
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(http.BAD_REQUEST,
url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
[docs] def test_update_all_image_existing_members(self):
"""
Test update existing image members
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
# Add several members to image
req = webob.Request.blank('/images/%s/members/test1' % UUID8)
req.method = 'PUT'
req.get_response(self.api)
fixture = [dict(member_id='test1', can_share=False)]
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(http.NO_CONTENT,
url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
[docs] def test_update_all_image_existing_deleted_members(self):
"""
Test update existing image members
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
# Add a new member to an image
req = webob.Request.blank('/images/%s/members/test1' % UUID8)
req.method = 'PUT'
req.get_response(self.api)
# Delete the existing member
self.get_api_response_ext(http.NO_CONTENT, method='DELETE',
url='/images/%s/members/test1' % UUID8)
# Re-add the deleted member by replacing membership list
fixture = [dict(member_id='test1', can_share=False)]
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(http.NO_CONTENT,
url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
memb_list = db_api.image_member_find(self.context, image_id=UUID8)
self.assertEqual(1, len(memb_list))
[docs] def test_add_member(self):
"""
Tests adding image members raises right exception
"""
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=False)
self.get_api_response_ext(http.UNAUTHORIZED, method='PUT',
url=('/images/%s/members/pattieblack' %
UUID2))
[docs] def test_add_member_to_image_positive(self):
"""
Test check that member can be successfully added
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_positive'
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(http.NO_CONTENT, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
[docs] def test_add_member_to_non_exist_image(self):
"""
Test check that member can't be added for
non exist image
"""
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_positive'
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(http.NOT_FOUND, url=test_uri % _gen_uuid(),
method='PUT', body=body,
content_type='json')
[docs] def test_add_image_member_non_shared_image_forbidden(self):
"""
Test update image members raises right exception
"""
test_rserver_api = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(
test_rserver_api, is_admin=False)
UUID9 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID9, size=19, protected=False)
db_api.image_create(self.context, extra_fixture)
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_to_non_share_image'
req = webob.Request.blank(test_uri % UUID9)
req.headers['X-Auth-Token'] = 'test1:test1:'
req.method = 'PUT'
req.content_type = 'application/json'
req.body = jsonutils.dump_as_bytes(dict(member=fixture))
res = req.get_response(api)
self.assertEqual(http.FORBIDDEN, res.status_int)
[docs] def test_add_member_to_image_bad_request(self):
"""
Test check right status code is returned
"""
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
fixture = [dict(can_share=True)]
test_uri = '/images/%s/members/test_add_member_bad_request'
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(http.BAD_REQUEST, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
[docs] def test_delete_member(self):
"""
Tests deleting image members raises right exception
"""
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=False)
self.get_api_response_ext(http.UNAUTHORIZED, method='DELETE',
url=('/images/%s/members/pattieblack' %
UUID2))
[docs] def test_delete_member_invalid(self):
"""
Tests deleting a invalid/non existing member raises right exception
"""
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=True)
res = self.get_api_response_ext(
http.NOT_FOUND, method='DELETE',
url=('/images/%s/members/pattieblack' % UUID2))
self.assertIn(b'Membership could not be found', res.body)
[docs] def test_delete_member_from_non_exist_image(self):
"""
Tests deleting image members raises right exception
"""
test_rserver_api = rserver.API(self.mapper)
self.api = test_utils.FakeAuthMiddleware(
test_rserver_api, is_admin=True)
test_uri = '/images/%s/members/pattieblack'
self.get_api_response_ext(http.NOT_FOUND, method='DELETE',
url=test_uri % _gen_uuid())
[docs] def test_delete_image_member_non_shared_image_forbidden(self):
"""
Test delete image members raises right exception
"""
test_rserver_api = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(
test_rserver_api, is_admin=False)
UUID9 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID9, size=19, protected=False)
db_api.image_create(self.context, extra_fixture)
test_uri = '/images/%s/members/test_add_member_to_non_share_image'
req = webob.Request.blank(test_uri % UUID9)
req.headers['X-Auth-Token'] = 'test1:test1:'
req.method = 'DELETE'
req.content_type = 'application/json'
res = req.get_response(api)
self.assertEqual(http.FORBIDDEN, res.status_int)
[docs] def test_add_member_delete_create(self):
"""
Test check that the same member can be successfully added after delete
it, and the same record will be reused for the same membership.
"""
# add a member
UUID8 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID8, size=19, protected=False,
owner='test user')
db_api.image_create(self.context, extra_fixture)
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_delete_create'
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(http.NO_CONTENT, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
memb_list = db_api.image_member_find(self.context, image_id=UUID8)
self.assertEqual(1, len(memb_list))
memb_list2 = db_api.image_member_find(self.context,
image_id=UUID8,
include_deleted=True)
self.assertEqual(1, len(memb_list2))
# delete the member
self.get_api_response_ext(http.NO_CONTENT, method='DELETE',
url=test_uri % UUID8)
memb_list = db_api.image_member_find(self.context, image_id=UUID8)
self.assertEqual(0, len(memb_list))
memb_list2 = db_api.image_member_find(self.context,
image_id=UUID8,
include_deleted=True)
self.assertEqual(1, len(memb_list2))
# create it again
self.get_api_response_ext(http.NO_CONTENT, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
memb_list = db_api.image_member_find(self.context, image_id=UUID8)
self.assertEqual(1, len(memb_list))
memb_list2 = db_api.image_member_find(self.context,
image_id=UUID8,
include_deleted=True)
self.assertEqual(1, len(memb_list2))
[docs] def test_get_on_image_member(self):
"""
Test GET on image members raises 404 and produces correct Allow headers
"""
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=False)
uri = '/images/%s/members/123' % UUID1
req = webob.Request.blank(uri)
req.method = 'GET'
res = req.get_response(self.api)
self.assertEqual(http.METHOD_NOT_ALLOWED, res.status_int)
self.assertIn(('Allow', 'PUT, DELETE'), res.headerlist)
[docs] def test_get_images_bad_urls(self):
"""Check that routes collections are not on (LP bug 1185828)"""
self.get_api_response_ext(http.NOT_FOUND, url='/images/detail.xxx')
self.get_api_response_ext(http.NOT_FOUND, url='/images.xxx')
self.get_api_response_ext(http.NOT_FOUND, url='/images/new')
self.get_api_response_ext(http.OK, url='/images/%s/members' % UUID1)
self.get_api_response_ext(http.NOT_FOUND,
url='/images/%s/members.xxx' % UUID1)
[docs]class TestRegistryAPILocations(base.IsolatedUnitTest,
test_utils.RegistryAPIMixIn):
[docs] def setUp(self):
"""Establish a clean test environment"""
super(TestRegistryAPILocations, self).setUp()
self.mapper = routes.Mapper()
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=True)
def _get_extra_fixture(id, name, **kwargs):
return self.get_extra_fixture(
id, name,
locations=[{'url': "file:///%s/%s" % (self.test_dir, id),
'metadata': {}, 'status': 'active'}], **kwargs)
self.FIXTURES = [
_get_extra_fixture(UUID1, 'fake image #1', is_public=False,
disk_format='ami', container_format='ami',
min_disk=0, min_ram=0, owner=123,
size=13, properties={'type': 'kernel'}),
_get_extra_fixture(UUID2, 'fake image #2',
min_disk=5, min_ram=256,
size=19, properties={})]
self.context = context.RequestContext(is_admin=True)
db_api.get_engine()
self.destroy_fixtures()
self.create_fixtures()
[docs] def tearDown(self):
"""Clear the test environment"""
super(TestRegistryAPILocations, self).tearDown()
self.destroy_fixtures()
[docs] def test_show_from_locations(self):
req = webob.Request.blank('/images/%s' % UUID1)
res = req.get_response(self.api)
self.assertEqual(http.OK, res.status_int)
res_dict = jsonutils.loads(res.body)
image = res_dict['image']
self.assertIn('id', image['location_data'][0])
image['location_data'][0].pop('id')
self.assertEqual(self.FIXTURES[0]['locations'][0],
image['location_data'][0])
self.assertEqual(self.FIXTURES[0]['locations'][0]['url'],
image['location_data'][0]['url'])
self.assertEqual(self.FIXTURES[0]['locations'][0]['metadata'],
image['location_data'][0]['metadata'])
[docs] def test_show_from_location_data(self):
req = webob.Request.blank('/images/%s' % UUID2)
res = req.get_response(self.api)
self.assertEqual(http.OK, res.status_int)
res_dict = jsonutils.loads(res.body)
image = res_dict['image']
self.assertIn('id', image['location_data'][0])
image['location_data'][0].pop('id')
self.assertEqual(self.FIXTURES[1]['locations'][0],
image['location_data'][0])
self.assertEqual(self.FIXTURES[1]['locations'][0]['url'],
image['location_data'][0]['url'])
self.assertEqual(self.FIXTURES[1]['locations'][0]['metadata'],
image['location_data'][0]['metadata'])
[docs] def test_create_from_location_data_with_encryption(self):
encryption_key = '1234567890123456'
location_url1 = "file:///%s/%s" % (self.test_dir, _gen_uuid())
location_url2 = "file:///%s/%s" % (self.test_dir, _gen_uuid())
encrypted_location_url1 = crypt.urlsafe_encrypt(encryption_key,
location_url1, 64)
encrypted_location_url2 = crypt.urlsafe_encrypt(encryption_key,
location_url2, 64)
fixture = {'name': 'fake image #3',
'status': 'active',
'disk_format': 'vhd',
'container_format': 'ovf',
'is_public': True,
'checksum': None,
'min_disk': 5,
'min_ram': 256,
'size': 19,
'location': encrypted_location_url1,
'location_data': [{'url': encrypted_location_url1,
'metadata': {'key': 'value'},
'status': 'active'},
{'url': encrypted_location_url2,
'metadata': {'key': 'value'},
'status': 'active'}]}
self.config(metadata_encryption_key=encryption_key)
req = webob.Request.blank('/images')
req.method = 'POST'
req.content_type = 'application/json'
req.body = jsonutils.dump_as_bytes(dict(image=fixture))
res = req.get_response(self.api)
self.assertEqual(http.OK, res.status_int)
res_dict = jsonutils.loads(res.body)
image = res_dict['image']
# NOTE(zhiyan) _normalize_image_location_for_db() function will
# not re-encrypted the url within location.
self.assertEqual(fixture['location'], image['location'])
self.assertEqual(2, len(image['location_data']))
self.assertEqual(fixture['location_data'][0]['url'],
image['location_data'][0]['url'])
self.assertEqual(fixture['location_data'][0]['metadata'],
image['location_data'][0]['metadata'])
self.assertEqual(fixture['location_data'][1]['url'],
image['location_data'][1]['url'])
self.assertEqual(fixture['location_data'][1]['metadata'],
image['location_data'][1]['metadata'])
image_entry = db_api.image_get(self.context, image['id'])
self.assertEqual(encrypted_location_url1,
image_entry['locations'][0]['url'])
self.assertEqual(encrypted_location_url2,
image_entry['locations'][1]['url'])
decrypted_location_url1 = crypt.urlsafe_decrypt(
encryption_key, image_entry['locations'][0]['url'])
decrypted_location_url2 = crypt.urlsafe_decrypt(
encryption_key, image_entry['locations'][1]['url'])
self.assertEqual(location_url1, decrypted_location_url1)
self.assertEqual(location_url2, decrypted_location_url2)
[docs]class TestSharability(test_utils.BaseTestCase):
[docs] def setUp(self):
super(TestSharability, self).setUp()
self.setup_db()
self.controller = glance.registry.api.v1.members.Controller()
[docs] def setup_db(self):
db_api.get_engine()
db_models.unregister_models(db_api.get_engine())
db_models.register_models(db_api.get_engine())
[docs] def test_is_image_sharable_as_admin(self):
TENANT1 = str(uuid.uuid4())
TENANT2 = str(uuid.uuid4())
ctxt1 = context.RequestContext(is_admin=False, tenant=TENANT1,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
ctxt2 = context.RequestContext(is_admin=True, user=TENANT2,
auth_token='user:%s:admin' % TENANT2,
owner_is_tenant=False)
UUIDX = str(uuid.uuid4())
# We need private image and context.owner should not match image
# owner
image = db_api.image_create(ctxt1, {'id': UUIDX,
'status': 'queued',
'is_public': False,
'owner': TENANT1})
result = self.controller.is_image_sharable(ctxt2, image)
self.assertTrue(result)
[docs] def test_is_image_sharable_owner_can_share(self):
TENANT1 = str(uuid.uuid4())
ctxt1 = context.RequestContext(is_admin=False, tenant=TENANT1,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
UUIDX = str(uuid.uuid4())
# We need private image and context.owner should not match image
# owner
image = db_api.image_create(ctxt1, {'id': UUIDX,
'status': 'queued',
'is_public': False,
'owner': TENANT1})
result = self.controller.is_image_sharable(ctxt1, image)
self.assertTrue(result)
[docs] def test_is_image_sharable_non_owner_cannot_share(self):
TENANT1 = str(uuid.uuid4())
TENANT2 = str(uuid.uuid4())
ctxt1 = context.RequestContext(is_admin=False, tenant=TENANT1,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
ctxt2 = context.RequestContext(is_admin=False, user=TENANT2,
auth_token='user:%s:user' % TENANT2,
owner_is_tenant=False)
UUIDX = str(uuid.uuid4())
# We need private image and context.owner should not match image
# owner
image = db_api.image_create(ctxt1, {'id': UUIDX,
'status': 'queued',
'is_public': False,
'owner': TENANT1})
result = self.controller.is_image_sharable(ctxt2, image)
self.assertFalse(result)
[docs] def test_is_image_sharable_non_owner_can_share_as_image_member(self):
TENANT1 = str(uuid.uuid4())
TENANT2 = str(uuid.uuid4())
ctxt1 = context.RequestContext(is_admin=False, tenant=TENANT1,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
ctxt2 = context.RequestContext(is_admin=False, user=TENANT2,
auth_token='user:%s:user' % TENANT2,
owner_is_tenant=False)
UUIDX = str(uuid.uuid4())
# We need private image and context.owner should not match image
# owner
image = db_api.image_create(ctxt1, {'id': UUIDX,
'status': 'queued',
'is_public': False,
'owner': TENANT1})
membership = {'can_share': True,
'member': TENANT2,
'image_id': UUIDX}
db_api.image_member_create(ctxt1, membership)
result = self.controller.is_image_sharable(ctxt2, image)
self.assertTrue(result)
[docs] def test_is_image_sharable_non_owner_as_image_member_without_sharing(self):
TENANT1 = str(uuid.uuid4())
TENANT2 = str(uuid.uuid4())
ctxt1 = context.RequestContext(is_admin=False, tenant=TENANT1,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
ctxt2 = context.RequestContext(is_admin=False, user=TENANT2,
auth_token='user:%s:user' % TENANT2,
owner_is_tenant=False)
UUIDX = str(uuid.uuid4())
# We need private image and context.owner should not match image
# owner
image = db_api.image_create(ctxt1, {'id': UUIDX,
'status': 'queued',
'is_public': False,
'owner': TENANT1})
membership = {'can_share': False,
'member': TENANT2,
'image_id': UUIDX}
db_api.image_member_create(ctxt1, membership)
result = self.controller.is_image_sharable(ctxt2, image)
self.assertFalse(result)
[docs] def test_is_image_sharable_owner_is_none(self):
TENANT1 = str(uuid.uuid4())
ctxt1 = context.RequestContext(is_admin=False, tenant=TENANT1,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
ctxt2 = context.RequestContext(is_admin=False, tenant=None,
auth_token='user:%s:user' % TENANT1,
owner_is_tenant=True)
UUIDX = str(uuid.uuid4())
# We need private image and context.owner should not match image
# owner
image = db_api.image_create(ctxt1, {'id': UUIDX,
'status': 'queued',
'is_public': False,
'owner': TENANT1})
result = self.controller.is_image_sharable(ctxt2, image)
self.assertFalse(result)