Merge "Remove Images API v1 entry points"

This commit is contained in:
Zuul 2018-07-27 06:08:31 +00:00 committed by Gerrit Code Review
commit ff77f59bd4
28 changed files with 21 additions and 6637 deletions

View File

@ -20,8 +20,6 @@ CONF = cfg.CONF
def root_app_factory(loader, global_conf, **local_conf):
if not CONF.enable_v1_api and '/v1' in local_conf:
del local_conf['/v1']
if not CONF.enable_v2_api and '/v2' in local_conf:
del local_conf['/v2']
return paste.urlmap.urlmap_factory(loader, global_conf, **local_conf)

View File

@ -31,7 +31,6 @@ import webob
from glance.api.common import size_checked_iter
from glance.api import policy
from glance.api.v1 import images
from glance.common import exception
from glance.common import utils
from glance.common import wsgi
@ -55,7 +54,6 @@ class CacheFilter(wsgi.Middleware):
def __init__(self, app):
self.cache = image_cache.ImageCache()
self.serializer = images.ImageSerializer()
self.policy = policy.Enforcer()
LOG.info(_LI("Initialized image cache middleware"))
super(CacheFilter, self).__init__(app)
@ -214,21 +212,6 @@ class CacheFilter(wsgi.Middleware):
else:
return (image_id, method, version)
def _process_v1_request(self, request, image_id, image_iterator,
image_meta):
# Don't display location
if 'location' in image_meta:
del image_meta['location']
image_meta.pop('location_data', None)
self._verify_metadata(image_meta)
response = webob.Response(request=request)
raw_response = {
'image_iterator': image_iterator,
'image_meta': image_meta,
}
return self.serializer.show(response, raw_response)
def _process_v2_request(self, request, image_id, image_iterator,
image_meta):
# We do some contortions to get the image_metadata so

View File

@ -72,10 +72,6 @@ class VersionNegotiationFilter(wsgi.Middleware):
def _get_allowed_versions(self):
allowed_versions = {}
if CONF.enable_v1_api:
allowed_versions['v1'] = 1
allowed_versions['v1.0'] = 1
allowed_versions['v1.1'] = 1
if CONF.enable_v2_api:
allowed_versions['v2'] = 2
allowed_versions['v2.0'] = 2

File diff suppressed because it is too large Load Diff

View File

@ -1,248 +0,0 @@
# Copyright 2012 OpenStack Foundation.
# Copyright 2013 NTT corp.
# 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.
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import encodeutils
import webob.exc
from glance.api import policy
from glance.api.v1 import controller
from glance.common import exception
from glance.common import utils
from glance.common import wsgi
from glance.i18n import _
import glance.registry.client.v1.api as registry
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
CONF.import_opt('image_member_quota', 'glance.common.config')
class Controller(controller.BaseController):
def __init__(self):
self.policy = policy.Enforcer()
def _check_can_access_image_members(self, context):
if context.owner is None and not context.is_admin:
raise webob.exc.HTTPUnauthorized(_("No authenticated user"))
def _enforce(self, req, action):
"""Authorize an action against our policies"""
try:
self.policy.enforce(req.context, action, {})
except exception.Forbidden:
LOG.debug("User not permitted to perform '%s' action", action)
raise webob.exc.HTTPForbidden()
def _raise_404_if_image_deleted(self, req, image_id):
image = self.get_image_meta_or_404(req, image_id)
if image['status'] == 'deleted':
msg = _("Image with identifier %s has been deleted.") % image_id
raise webob.exc.HTTPNotFound(msg)
def index(self, req, image_id):
"""
Return a list of dictionaries indicating the members of the
image, i.e., those tenants the image is shared with.
:param req: the Request object coming from the wsgi layer
:param image_id: The opaque image identifier
:returns: The response body is a mapping of the following form
::
{'members': [
{'member_id': <MEMBER>,
'can_share': <SHARE_PERMISSION>, ...}, ...
]}
"""
self._enforce(req, 'get_members')
self._raise_404_if_image_deleted(req, image_id)
try:
members = registry.get_image_members(req.context, image_id)
except exception.NotFound:
msg = _("Image with identifier %s not found") % image_id
LOG.warn(msg)
raise webob.exc.HTTPNotFound(msg)
except exception.Forbidden:
msg = _("Unauthorized image access")
LOG.warn(msg)
raise webob.exc.HTTPForbidden(msg)
return dict(members=members)
@utils.mutating
def delete(self, req, image_id, id):
"""
Removes a membership from the image.
"""
self._check_can_access_image_members(req.context)
self._enforce(req, 'delete_member')
self._raise_404_if_image_deleted(req, image_id)
try:
registry.delete_member(req.context, image_id, id)
self._update_store_acls(req, image_id)
except exception.NotFound as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Forbidden as e:
LOG.debug("User not permitted to remove membership from image "
"'%s'", image_id)
raise webob.exc.HTTPNotFound(explanation=e.msg)
return webob.exc.HTTPNoContent()
def default(self, req, image_id, id, body=None):
"""This will cover the missing 'show' and 'create' actions"""
raise webob.exc.HTTPMethodNotAllowed()
def _enforce_image_member_quota(self, req, attempted):
if CONF.image_member_quota < 0:
# If value is negative, allow unlimited number of members
return
maximum = CONF.image_member_quota
if attempted > maximum:
msg = _("The limit has been exceeded on the number of allowed "
"image members for this image. Attempted: %(attempted)s, "
"Maximum: %(maximum)s") % {'attempted': attempted,
'maximum': maximum}
raise webob.exc.HTTPRequestEntityTooLarge(explanation=msg,
request=req)
@utils.mutating
def update(self, req, image_id, id, body=None):
"""
Adds a membership to the image, or updates an existing one.
If a body is present, it is a dict with the following format
::
{'member': {
'can_share': [True|False]
}}
If `can_share` is provided, the member's ability to share is
set accordingly. If it is not provided, existing memberships
remain unchanged and new memberships default to False.
"""
self._check_can_access_image_members(req.context)
self._enforce(req, 'modify_member')
self._raise_404_if_image_deleted(req, image_id)
new_number_of_members = len(registry.get_image_members(req.context,
image_id)) + 1
self._enforce_image_member_quota(req, new_number_of_members)
# Figure out can_share
can_share = None
if body and 'member' in body and 'can_share' in body['member']:
can_share = bool(body['member']['can_share'])
try:
registry.add_member(req.context, image_id, id, can_share)
self._update_store_acls(req, image_id)
except exception.Invalid as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPBadRequest(explanation=e.msg)
except exception.NotFound as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Forbidden as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPNotFound(explanation=e.msg)
return webob.exc.HTTPNoContent()
@utils.mutating
def update_all(self, req, image_id, body):
"""
Replaces the members of the image with those specified in the
body. The body is a dict with the following format
::
{'memberships': [
{'member_id': <MEMBER_ID>,
['can_share': [True|False]]}, ...
]}
"""
self._check_can_access_image_members(req.context)
self._enforce(req, 'modify_member')
self._raise_404_if_image_deleted(req, image_id)
memberships = body.get('memberships')
if memberships:
new_number_of_members = len(body['memberships'])
self._enforce_image_member_quota(req, new_number_of_members)
try:
registry.replace_members(req.context, image_id, body)
self._update_store_acls(req, image_id)
except exception.Invalid as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPBadRequest(explanation=e.msg)
except exception.NotFound as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Forbidden as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPNotFound(explanation=e.msg)
return webob.exc.HTTPNoContent()
def index_shared_images(self, req, id):
"""
Retrieves list of image memberships for the given member.
:param req: the Request object coming from the wsgi layer
:param id: the opaque member identifier
:returns: The response body is a mapping of the following form
::
{'shared_images': [
{'image_id': <IMAGE>,
'can_share': <SHARE_PERMISSION>, ...}, ...
]}
"""
try:
members = registry.get_member_images(req.context, id)
except exception.NotFound as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Forbidden as e:
LOG.debug(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPForbidden(explanation=e.msg)
return dict(shared_images=members)
def _update_store_acls(self, req, image_id):
image_meta = self.get_image_meta_or_404(req, image_id)
location_uri = image_meta.get('location')
public = image_meta.get('is_public')
self.update_store_acls(req, image_id, location_uri, public)
def create_resource():
"""Image members resource factory method"""
deserializer = wsgi.JSONRequestDeserializer()
serializer = wsgi.JSONResponseSerializer()
return wsgi.Resource(Controller(), deserializer, serializer)

View File

@ -14,8 +14,6 @@
# under the License.
from glance.api.v1 import images
from glance.api.v1 import members
from glance.common import wsgi
@ -26,84 +24,8 @@ class API(wsgi.Router):
def __init__(self, mapper):
reject_method_resource = wsgi.Resource(wsgi.RejectMethodController())
images_resource = images.create_resource()
mapper.connect("/",
controller=images_resource,
action="index")
mapper.connect("/images",
controller=images_resource,
action='index',
conditions={'method': ['GET']})
mapper.connect("/images",
controller=images_resource,
action='create',
conditions={'method': ['POST']})
mapper.connect("/images",
controller=reject_method_resource,
action='reject',
allowed_methods='GET, POST')
mapper.connect("/images/detail",
controller=images_resource,
action='detail',
conditions={'method': ['GET', 'HEAD']})
mapper.connect("/images/detail",
controller=reject_method_resource,
action='reject',
allowed_methods='GET, HEAD')
mapper.connect("/images/{id}",
controller=images_resource,
action="meta",
conditions=dict(method=["HEAD"]))
mapper.connect("/images/{id}",
controller=images_resource,
action="show",
conditions=dict(method=["GET"]))
mapper.connect("/images/{id}",
controller=images_resource,
action="update",
conditions=dict(method=["PUT"]))
mapper.connect("/images/{id}",
controller=images_resource,
action="delete",
conditions=dict(method=["DELETE"]))
mapper.connect("/images/{id}",
controller=reject_method_resource,
action='reject',
allowed_methods='GET, HEAD, PUT, DELETE')
members_resource = members.create_resource()
mapper.connect("/images/{image_id}/members",
controller=members_resource,
action="index",
conditions={'method': ['GET']})
mapper.connect("/images/{image_id}/members",
controller=members_resource,
action="update_all",
conditions=dict(method=["PUT"]))
mapper.connect("/images/{image_id}/members",
controller=reject_method_resource,
action='reject',
allowed_methods='GET, PUT')
mapper.connect("/images/{image_id}/members/{id}",
controller=members_resource,
action="show",
conditions={'method': ['GET']})
mapper.connect("/images/{image_id}/members/{id}",
controller=members_resource,
action="update",
conditions={'method': ['PUT']})
mapper.connect("/images/{image_id}/members/{id}",
controller=members_resource,
action="delete",
conditions={'method': ['DELETE']})
mapper.connect("/images/{image_id}/members/{id}",
controller=reject_method_resource,
action='reject',
allowed_methods='GET, PUT, DELETE')
mapper.connect("/shared-images/{id}",
controller=members_resource,
action="index_shared_images")
action="reject")
super(API, self).__init__(mapper)

View File

@ -20,7 +20,7 @@ from six.moves import http_client
import webob.dec
from glance.common import wsgi
from glance.i18n import _, _LW
from glance.i18n import _
versions_opts = [
@ -82,20 +82,6 @@ class Controller(object):
build_version_object(2.1, 'v2', 'SUPPORTED'),
build_version_object(2.0, 'v2', 'SUPPORTED'),
])
if CONF.enable_v1_api:
LOG.warn(_LW('The Images (Glance) v1 API is deprecated and will '
'be removed on or after the Pike release, following '
'the standard OpenStack deprecation policy. '
'Currently, the solution is to set '
'enable_v1_api=False and enable_v2_api=True in your '
'glance-api.conf file. Once those options are '
'removed from the code, Images (Glance) v2 API will '
'be switched on by default and will be the only '
'option to deploy and use.'))
version_objs.extend([
build_version_object(1.1, 'v1', 'DEPRECATED'),
build_version_object(1.0, 'v1', 'DEPRECATED'),
])
status = explicit and http_client.OK or http_client.MULTIPLE_CHOICES
response = webob.Response(request=req,

View File

@ -1,490 +0,0 @@
#!/usr/bin/env python
# Copyright 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.
"""
A simple cache management utility for Glance.
"""
from __future__ import print_function
import argparse
import collections
import datetime
import functools
import os
import sys
import time
from oslo_utils import encodeutils
import prettytable
from six.moves import input
# If ../glance/__init__.py exists, add ../ to Python search path, so that
# it will override what happens to be installed in /usr/(local/)lib/python...
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir, 'glance', '__init__.py')):
sys.path.insert(0, possible_topdir)
from glance.common import exception
import glance.image_cache.client
from glance.version import version_info as version
SUCCESS = 0
FAILURE = 1
def catch_error(action):
"""Decorator to provide sensible default error handling for actions."""
def wrap(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
ret = func(*args, **kwargs)
return SUCCESS if ret is None else ret
except exception.NotFound:
options = args[0]
print("Cache management middleware not enabled on host %s" %
options.host)
return FAILURE
except exception.Forbidden:
print("Not authorized to make this request.")
return FAILURE
except Exception as e:
options = args[0]
if options.debug:
raise
print("Failed to %s. Got error:" % action)
pieces = encodeutils.exception_to_unicode(e).split('\n')
for piece in pieces:
print(piece)
return FAILURE
return wrapper
return wrap
@catch_error('show cached images')
def list_cached(args):
"""%(prog)s list-cached [options]
List all images currently cached.
"""
client = get_client(args)
images = client.get_cached_images()
if not images:
print("No cached images.")
return SUCCESS
print("Found %d cached images..." % len(images))
pretty_table = prettytable.PrettyTable(("ID",
"Last Accessed (UTC)",
"Last Modified (UTC)",
"Size",
"Hits"))
pretty_table.align['Size'] = "r"
pretty_table.align['Hits'] = "r"
for image in images:
last_accessed = image['last_accessed']
if last_accessed == 0:
last_accessed = "N/A"
else:
last_accessed = datetime.datetime.utcfromtimestamp(
last_accessed).isoformat()
pretty_table.add_row((
image['image_id'],
last_accessed,
datetime.datetime.utcfromtimestamp(
image['last_modified']).isoformat(),
image['size'],
image['hits']))
print(pretty_table.get_string())
return SUCCESS
@catch_error('show queued images')
def list_queued(args):
"""%(prog)s list-queued [options]
List all images currently queued for caching.
"""
client = get_client(args)
images = client.get_queued_images()
if not images:
print("No queued images.")
return SUCCESS
print("Found %d queued images..." % len(images))
pretty_table = prettytable.PrettyTable(("ID",))
for image in images:
pretty_table.add_row((image,))
print(pretty_table.get_string())
@catch_error('queue the specified image for caching')
def queue_image(args):
"""%(prog)s queue-image <IMAGE_ID> [options]
Queues an image for caching.
"""
if len(args.command) == 2:
image_id = args.command[1]
else:
print("Please specify one and only ID of the image you wish to ")
print("queue from the cache as the first argument")
return FAILURE
if (not args.force and
not user_confirm("Queue image %(image_id)s for caching?" %
{'image_id': image_id}, default=False)):
return SUCCESS
client = get_client(args)
client.queue_image_for_caching(image_id)
if args.verbose:
print("Queued image %(image_id)s for caching" %
{'image_id': image_id})
return SUCCESS
@catch_error('delete the specified cached image')
def delete_cached_image(args):
"""%(prog)s delete-cached-image <IMAGE_ID> [options]
Deletes an image from the cache.
"""
if len(args.command) == 2:
image_id = args.command[1]
else:
print("Please specify one and only ID of the image you wish to ")
print("delete from the cache as the first argument")
return FAILURE
if (not args.force and
not user_confirm("Delete cached image %(image_id)s?" %
{'image_id': image_id}, default=False)):
return SUCCESS
client = get_client(args)
client.delete_cached_image(image_id)
if args.verbose:
print("Deleted cached image %(image_id)s" % {'image_id': image_id})
return SUCCESS
@catch_error('Delete all cached images')
def delete_all_cached_images(args):
"""%(prog)s delete-all-cached-images [options]
Remove all images from the cache.
"""
if (not args.force and
not user_confirm("Delete all cached images?", default=False)):
return SUCCESS
client = get_client(args)
num_deleted = client.delete_all_cached_images()
if args.verbose:
print("Deleted %(num_deleted)s cached images" %
{'num_deleted': num_deleted})
return SUCCESS
@catch_error('delete the specified queued image')
def delete_queued_image(args):
"""%(prog)s delete-queued-image <IMAGE_ID> [options]
Deletes an image from the cache.
"""
if len(args.command) == 2:
image_id = args.command[1]
else:
print("Please specify one and only ID of the image you wish to ")
print("delete from the cache as the first argument")
return FAILURE
if (not args.force and
not user_confirm("Delete queued image %(image_id)s?" %
{'image_id': image_id}, default=False)):
return SUCCESS
client = get_client(args)
client.delete_queued_image(image_id)
if args.verbose:
print("Deleted queued image %(image_id)s" % {'image_id': image_id})
return SUCCESS
@catch_error('Delete all queued images')
def delete_all_queued_images(args):
"""%(prog)s delete-all-queued-images [options]
Remove all images from the cache queue.
"""
if (not args.force and
not user_confirm("Delete all queued images?", default=False)):
return SUCCESS
client = get_client(args)
num_deleted = client.delete_all_queued_images()
if args.verbose:
print("Deleted %(num_deleted)s queued images" %
{'num_deleted': num_deleted})
return SUCCESS
def get_client(options):
"""Return a new client object to a Glance server.
specified by the --host and --port options
supplied to the CLI
"""
return glance.image_cache.client.get_client(
host=options.host,
port=options.port,
username=options.os_username,
password=options.os_password,
tenant=options.os_tenant_name,
auth_url=options.os_auth_url,
auth_strategy=options.os_auth_strategy,
auth_token=options.os_auth_token,
region=options.os_region_name,
insecure=options.insecure)
def env(*vars, **kwargs):
"""Search for the first defined of possibly many env vars.
Returns the first environment variable defined in vars, or
returns the default defined in kwargs.
"""
for v in vars:
value = os.environ.get(v)
if value:
return value
return kwargs.get('default', '')
def print_help(args):
"""
Print help specific to a command
"""
command = lookup_command(args.command[1])
print(command.__doc__ % {'prog': os.path.basename(sys.argv[0])})
def parse_args(parser):
"""Set up the CLI and config-file options that may be
parsed and program commands.
:param parser: The option parser
"""
parser.add_argument('command', default='help', nargs='*',
help='The command to execute')
parser.add_argument('-v', '--verbose', default=False, action="store_true",
help="Print more verbose output.")
parser.add_argument('-d', '--debug', default=False, action="store_true",
help="Print debugging output.")
parser.add_argument('-H', '--host', metavar="ADDRESS", default="0.0.0.0",
help="Address of Glance API host.")
parser.add_argument('-p', '--port', dest="port", metavar="PORT",
type=int, default=9292,
help="Port the Glance API host listens on.")
parser.add_argument('-k', '--insecure', dest="insecure",
default=False, action="store_true",
help='Explicitly allow glance to perform "insecure" '
"SSL (https) requests. The server's certificate "
"will not be verified against any certificate "
"authorities. This option should be used with "
"caution.")
parser.add_argument('-f', '--force', dest="force",
default=False, action="store_true",
help="Prevent select actions from requesting "
"user confirmation.")
parser.add_argument('--os-auth-token',
dest='os_auth_token',
default=env('OS_AUTH_TOKEN'),
help='Defaults to env[OS_AUTH_TOKEN].')
parser.add_argument('-A', '--os_auth_token', '--auth_token',
dest='os_auth_token',
help=argparse.SUPPRESS)
parser.add_argument('--os-username',
dest='os_username',
default=env('OS_USERNAME'),
help='Defaults to env[OS_USERNAME].')
parser.add_argument('-I', '--os_username',
dest='os_username',
help=argparse.SUPPRESS)
parser.add_argument('--os-password',
dest='os_password',
default=env('OS_PASSWORD'),
help='Defaults to env[OS_PASSWORD].')
parser.add_argument('-K', '--os_password',
dest='os_password',
help=argparse.SUPPRESS)
parser.add_argument('--os-region-name',
dest='os_region_name',
default=env('OS_REGION_NAME'),
help='Defaults to env[OS_REGION_NAME].')
parser.add_argument('-R', '--os_region_name',
dest='os_region_name',
help=argparse.SUPPRESS)
parser.add_argument('--os-tenant-id',
dest='os_tenant_id',
default=env('OS_TENANT_ID'),
help='Defaults to env[OS_TENANT_ID].')
parser.add_argument('--os_tenant_id',
dest='os_tenant_id',
help=argparse.SUPPRESS)
parser.add_argument('--os-tenant-name',
dest='os_tenant_name',
default=env('OS_TENANT_NAME'),
help='Defaults to env[OS_TENANT_NAME].')
parser.add_argument('-T', '--os_tenant_name',
dest='os_tenant_name',
help=argparse.SUPPRESS)
parser.add_argument('--os-auth-url',
default=env('OS_AUTH_URL'),
help='Defaults to env[OS_AUTH_URL].')
parser.add_argument('-N', '--os_auth_url',
dest='os_auth_url',
help=argparse.SUPPRESS)
parser.add_argument('-S', '--os_auth_strategy', dest="os_auth_strategy",
metavar="STRATEGY",
help="Authentication strategy (keystone or noauth).")
version_string = version.cached_version_string()
parser.add_argument('--version', action='version',
version=version_string)
return parser.parse_args()
CACHE_COMMANDS = collections.OrderedDict()
CACHE_COMMANDS['help'] = (
print_help, 'Output help for one of the commands below')
CACHE_COMMANDS['list-cached'] = (
list_cached, 'List all images currently cached')
CACHE_COMMANDS['list-queued'] = (
list_queued, 'List all images currently queued for caching')
CACHE_COMMANDS['queue-image'] = (
queue_image, 'Queue an image for caching')
CACHE_COMMANDS['delete-cached-image'] = (
delete_cached_image, 'Purges an image from the cache')
CACHE_COMMANDS['delete-all-cached-images'] = (
delete_all_cached_images, 'Removes all images from the cache')
CACHE_COMMANDS['delete-queued-image'] = (
delete_queued_image, 'Deletes an image from the cache queue')
CACHE_COMMANDS['delete-all-queued-images'] = (
delete_all_queued_images, 'Deletes all images from the cache queue')
def _format_command_help():
"""Formats the help string for subcommands."""
help_msg = "Commands:\n\n"
for command, info in CACHE_COMMANDS.items():
if command == 'help':
command = 'help <command>'
help_msg += " %-28s%s\n\n" % (command, info[1])
return help_msg
def lookup_command(command_name):
try:
command = CACHE_COMMANDS[command_name]
return command[0]
except KeyError:
print('\nError: "%s" is not a valid command.\n' % command_name)
print(_format_command_help())
sys.exit("Unknown command: %(cmd_name)s" % {'cmd_name': command_name})
def user_confirm(prompt, default=False):
"""Yes/No question dialog with user.
:param prompt: question/statement to present to user (string)
:param default: boolean value to return if empty string
is received as response to prompt
"""
if default:
prompt_default = "[Y/n]"
else:
prompt_default = "[y/N]"
answer = input("%s %s " % (prompt, prompt_default))
if answer == "":
return default
else:
return answer.lower() in ("yes", "y")
def main():
parser = argparse.ArgumentParser(
description=_format_command_help(),
formatter_class=argparse.RawDescriptionHelpFormatter)
args = parse_args(parser)
if args.command[0] == 'help' and len(args.command) == 1:
parser.print_help()
return
# Look up the command to run
command = lookup_command(args.command[0])
try:
start_time = time.time()
result = command(args)
end_time = time.time()
if args.verbose:
print("Completed in %-0.4f sec." % (end_time - start_time))
sys.exit(result)
except (RuntimeError, NotImplementedError) as e:
sys.exit("ERROR: %s" % e)
if __name__ == '__main__':
main()

View File

@ -454,50 +454,6 @@ Possible values:
Related options:
* None
""")),
# NOTE(nikhil): Even though deprecated, the configuration option
# ``enable_v1_api`` is set to True by default on purpose. Having it enabled
# helps the projects that haven't been able to fully move to v2 yet by
# keeping the devstack setup to use glance v1 as well. We need to switch it
# to False by default soon after Newton is cut so that we can identify the
# projects that haven't moved to v2 yet and start having some interesting
# conversations with them. Switching to False in Newton may result into
# destabilizing the gate and affect the release.
cfg.BoolOpt('enable_v1_api',
default=True,
deprecated_reason=_DEPRECATE_GLANCE_V1_MSG,
deprecated_since='Newton',
help=_("""
Deploy the v1 OpenStack Images API.
When this option is set to ``True``, Glance service will respond to
requests on registered endpoints conforming to the v1 OpenStack
Images API.
NOTES:
* If this option is enabled, then ``enable_v1_registry`` must
also be set to ``True`` to enable mandatory usage of Registry
service with v1 API.
* If this option is disabled, then the ``enable_v1_registry``
option, which is enabled by default, is also recommended
to be disabled.
* This option is separate from ``enable_v2_api``, both v1 and v2
OpenStack Images API can be deployed independent of each
other.
* If deploying only the v2 Images API, this option, which is
enabled by default, should be disabled.
Possible values:
* True
* False
Related options:
* enable_v1_registry
* enable_v2_api
""")),
cfg.BoolOpt('enable_v2_api',
default=True,
@ -523,20 +479,12 @@ NOTES:
option, which is enabled by default, is also recommended
to be disabled.
* This option is separate from ``enable_v1_api``, both v1 and v2
OpenStack Images API can be deployed independent of each
other.
* If deploying only the v1 Images API, this option, which is
enabled by default, should be disabled.
Possible values:
* True
* False
Related options:
* enable_v2_registry
* enable_v1_api
""")),
cfg.BoolOpt('enable_v1_registry',
@ -544,25 +492,7 @@ Related options:
deprecated_reason=_DEPRECATE_GLANCE_V1_MSG,
deprecated_since='Newton',
help=_("""
Deploy the v1 API Registry service.
When this option is set to ``True``, the Registry service
will be enabled in Glance for v1 API requests.
NOTES:
* Use of Registry is mandatory in v1 API, so this option must
be set to ``True`` if the ``enable_v1_api`` option is enabled.
* If deploying only the v2 OpenStack Images API, this option,
which is enabled by default, should be disabled.
Possible values:
* True
* False
Related options:
* enable_v1_api
DEPRECATED FOR REMOVAL
""")),
cfg.BoolOpt('enable_v2_registry',
default=True,

View File

@ -27,6 +27,7 @@ from glance import scrubber
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
CONF.import_opt('use_user_token', 'glance.registry.client')
RESTRICTED_URI_SCHEMAS = frozenset(['file', 'filesystem', 'swift+config'])

View File

@ -1,132 +0,0 @@
# 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 os
from oslo_serialization import jsonutils as json
from glance.common import client as base_client
from glance.common import exception
from glance.i18n import _
class CacheClient(base_client.BaseClient):
DEFAULT_PORT = 9292
DEFAULT_DOC_ROOT = '/v1'
def delete_cached_image(self, image_id):
"""
Delete a specified image from the cache
"""
self.do_request("DELETE", "/cached_images/%s" % image_id)
return True
def get_cached_images(self, **kwargs):
"""
Returns a list of images stored in the image cache.
"""
res = self.do_request("GET", "/cached_images")
data = json.loads(res.read())['cached_images']
return data
def get_queued_images(self, **kwargs):
"""
Returns a list of images queued for caching
"""
res = self.do_request("GET", "/queued_images")
data = json.loads(res.read())['queued_images']
return data
def delete_all_cached_images(self):
"""
Delete all cached images
"""
res = self.do_request("DELETE", "/cached_images")
data = json.loads(res.read())
num_deleted = data['num_deleted']
return num_deleted
def queue_image_for_caching(self, image_id):
"""
Queue an image for prefetching into cache
"""
self.do_request("PUT", "/queued_images/%s" % image_id)
return True
def delete_queued_image(self, image_id):
"""
Delete a specified image from the cache queue
"""
self.do_request("DELETE", "/queued_images/%s" % image_id)
return True
def delete_all_queued_images(self):
"""
Delete all queued images
"""
res = self.do_request("DELETE", "/queued_images")
data = json.loads(res.read())
num_deleted = data['num_deleted']
return num_deleted
def get_client(host, port=None, timeout=None, use_ssl=False, username=None,
password=None, tenant=None,
auth_url=None, auth_strategy=None,
auth_token=None, region=None,
is_silent_upload=False, insecure=False):
"""
Returns a new client Glance client object based on common kwargs.
If an option isn't specified falls back to common environment variable
defaults.
"""
if auth_url or os.getenv('OS_AUTH_URL'):
force_strategy = 'keystone'
else:
force_strategy = None
creds = {
'username': username or
os.getenv('OS_AUTH_USER', os.getenv('OS_USERNAME')),
'password': password or
os.getenv('OS_AUTH_KEY', os.getenv('OS_PASSWORD')),
'tenant': tenant or
os.getenv('OS_AUTH_TENANT', os.getenv('OS_TENANT_NAME')),
'auth_url': auth_url or
os.getenv('OS_AUTH_URL'),
'strategy': force_strategy or
auth_strategy or
os.getenv('OS_AUTH_STRATEGY', 'noauth'),
'region': region or
os.getenv('OS_REGION_NAME'),
}
if creds['strategy'] == 'keystone' and not creds['auth_url']:
msg = _("--os_auth_url option or OS_AUTH_URL environment variable "
"required when keystone authentication strategy is enabled\n")
raise exception.ClientConfigurationError(msg)
return CacheClient(
host=host,
port=port,
timeout=timeout,
use_ssl=use_ssl,
auth_token=auth_token or
os.getenv('OS_TOKEN'),
creds=creds,
insecure=insecure,
configure_via_auth=False)

View File

@ -74,9 +74,7 @@ class Server(object):
self.show_image_direct_url = False
self.show_multiple_locations = False
self.property_protection_file = ''
self.enable_v1_api = True
self.enable_v2_api = True
self.enable_v1_registry = True
self.enable_v2_registry = True
self.needs_database = False
self.log_file = None
@ -346,7 +344,6 @@ sql_connection = %(sql_connection)s
show_image_direct_url = %(show_image_direct_url)s
show_multiple_locations = %(show_multiple_locations)s
user_storage_quota = %(user_storage_quota)s
enable_v1_api = %(enable_v1_api)s
enable_v2_api = %(enable_v2_api)s
lock_path = %(lock_path)s
property_protection_file = %(property_protection_file)s

View File

@ -78,8 +78,10 @@ class TestScrubber(functional.FunctionalTest):
scrubs them
"""
self.cleanup()
kwargs = self.__dict__.copy()
kwargs['use_user_token'] = True
self.start_servers(delayed_delete=True, daemon=True,
metadata_encryption_key='')
metadata_encryption_key='', **kwargs)
path = "http://%s:%d/v2/images" % ("127.0.0.1", self.api_port)
response, content = self._send_create_image_http_request(path)
self.assertEqual(http_client.CREATED, response.status)
@ -112,8 +114,10 @@ class TestScrubber(functional.FunctionalTest):
daemon mode
"""
self.cleanup()
kwargs = self.__dict__.copy()
kwargs['use_user_token'] = True
self.start_servers(delayed_delete=True, daemon=False,
metadata_encryption_key='')
metadata_encryption_key='', **kwargs)
path = "http://%s:%d/v2/images" % ("127.0.0.1", self.api_port)
response, content = self._send_create_image_http_request(path)
self.assertEqual(http_client.CREATED, response.status)
@ -159,8 +163,10 @@ class TestScrubber(functional.FunctionalTest):
# Start servers.
self.cleanup()
kwargs = self.__dict__.copy()
kwargs['use_user_token'] = True
self.start_servers(delayed_delete=True, daemon=False,
default_store='file')
default_store='file', **kwargs)
# Check that we are using a file backend.
self.assertEqual(self.api_server.default_store, 'file')
@ -235,8 +241,10 @@ class TestScrubber(functional.FunctionalTest):
def test_scrubber_restore_image(self):
self.cleanup()
kwargs = self.__dict__.copy()
kwargs['use_user_token'] = True
self.start_servers(delayed_delete=True, daemon=False,
metadata_encryption_key='')
metadata_encryption_key='', **kwargs)
path = "http://%s:%d/v2/images" % ("127.0.0.1", self.api_port)
response, content = self._send_create_image_http_request(path)
self.assertEqual(http_client.CREATED, response.status)

View File

@ -22,25 +22,6 @@ from six.moves import http_client
from glance.tests import functional
# TODO(rosmaita): all the EXPERIMENTAL stuff in this file can be ripped out
# when v2.6 becomes CURRENT in Queens
def _generate_v1_versions(url):
v1_versions = {'versions': [
{
'id': 'v1.1',
'status': 'DEPRECATED',
'links': [{'rel': 'self', 'href': url % '1'}],
},
{
'id': 'v1.0',
'status': 'DEPRECATED',
'links': [{'rel': 'self', 'href': url % '1'}],
},
]}
return v1_versions
def _generate_v2_versions(url):
version_list = []
@ -86,9 +67,8 @@ def _generate_v2_versions(url):
def _generate_all_versions(url):
v1 = _generate_v1_versions(url)
v2 = _generate_v2_versions(url)
all_versions = {'versions': v2['versions'] + v1['versions']}
all_versions = {'versions': v2['versions']}
return all_versions
@ -96,7 +76,6 @@ class TestApiVersions(functional.FunctionalTest):
def test_version_configurations(self):
"""Test that versioning is handled properly through all channels"""
# v1 and v2 api enabled
self.start_servers(**self.__dict__.copy())
url = 'http://127.0.0.1:%d/v%%s/' % self.api_port
@ -111,7 +90,6 @@ class TestApiVersions(functional.FunctionalTest):
self.assertEqual(versions, content)
def test_v2_api_configuration(self):
self.api_server.enable_v1_api = False
self.api_server.enable_v2_api = True
self.start_servers(**self.__dict__.copy())
@ -126,22 +104,6 @@ class TestApiVersions(functional.FunctionalTest):
content = jsonutils.loads(content_json.decode())
self.assertEqual(versions, content)
def test_v1_api_configuration(self):
self.api_server.enable_v1_api = True
self.api_server.enable_v2_api = False
self.start_servers(**self.__dict__.copy())
url = 'http://127.0.0.1:%d/v%%s/' % self.api_port
versions = _generate_v1_versions(url)
# Verify version choices returned.
path = 'http://%s:%d' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content_json = http.request(path, 'GET')
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(versions, content)
class TestApiPaths(functional.FunctionalTest):
def setUp(self):
@ -165,26 +127,6 @@ class TestApiPaths(functional.FunctionalTest):
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_images_path(self):
"""Assert GET /images with `no Accept:` header.
Verify version choices returned.
"""
path = 'http://%s:%d/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content_json = http.request(path, 'GET')
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_v1_images_path(self):
"""GET /v1/images with `no Accept:` header.
Verify empty images list returned.
"""
path = 'http://%s:%d/v1/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
def test_get_root_path_with_unknown_header(self):
"""Assert GET / with Accept: unknown header
Verify version choices returned. Verify message in API log about
@ -198,49 +140,6 @@ class TestApiPaths(functional.FunctionalTest):
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_root_path_with_openstack_header(self):
"""Assert GET / with an Accept: application/vnd.openstack.images-v1
Verify empty image list returned
"""
path = 'http://%s:%d/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
headers = {'Accept': 'application/vnd.openstack.images-v1'}
response, content = http.request(path, 'GET', headers=headers)
self.assertEqual(http_client.OK, response.status)
self.assertEqual(self.images_json, content.decode())
def test_get_images_path_with_openstack_header(self):
"""Assert GET /images with a
`Accept: application/vnd.openstack.compute-v1` header.
Verify version choices returned. Verify message in API log
about unknown accept header.
"""
path = 'http://%s:%d/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
headers = {'Accept': 'application/vnd.openstack.compute-v1'}
response, content_json = http.request(path, 'GET', headers=headers)
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_v10_images_path(self):
"""Assert GET /v1.0/images with no Accept: header
Verify version choices returned
"""
path = 'http://%s:%d/v1.a/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
def test_get_v1a_images_path(self):
"""Assert GET /v1.a/images with no Accept: header
Verify version choices returned
"""
path = 'http://%s:%d/v1.a/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
def test_get_va1_images_path(self):
"""Assert GET /va.1/images with no Accept: header
Verify version choices returned
@ -263,28 +162,6 @@ class TestApiPaths(functional.FunctionalTest):
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_versions_path_with_openstack_header(self):
"""Assert GET /versions with the
`Accept: application/vnd.openstack.images-v1` header.
Verify version choices returned.
"""
path = 'http://%s:%d/versions' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
headers = {'Accept': 'application/vnd.openstack.images-v1'}
response, content_json = http.request(path, 'GET', headers=headers)
self.assertEqual(http_client.OK, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_v1_versions_path(self):
"""Assert GET /v1/versions with `no Accept:` header
Verify 404 returned
"""
path = 'http://%s:%d/v1/versions' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.NOT_FOUND, response.status)
def test_get_versions_choices(self):
"""Verify version choices returned"""
path = 'http://%s:%d/v10' % ('127.0.0.1', self.api_port)
@ -293,28 +170,3 @@ class TestApiPaths(functional.FunctionalTest):
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_images_path_with_openstack_v2_header(self):
"""Assert GET /images with a
`Accept: application/vnd.openstack.compute-v2` header.
Verify version choices returned. Verify message in API log
about unknown version in accept header.
"""
path = 'http://%s:%d/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
headers = {'Accept': 'application/vnd.openstack.images-v10'}
response, content_json = http.request(path, 'GET', headers=headers)
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)
def test_get_v12_images_path(self):
"""Assert GET /v1.2/images with `no Accept:` header
Verify version choices returned
"""
path = 'http://%s:%d/v1.2/images' % ('127.0.0.1', self.api_port)
http = httplib2.Http()
response, content_json = http.request(path, 'GET')
self.assertEqual(http_client.MULTIPLE_CHOICES, response.status)
content = jsonutils.loads(content_json.decode())
self.assertEqual(self.versions, content)

View File

@ -1,358 +0,0 @@
# Copyright 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.
"""Functional test case that utilizes the bin/glance-cache-manage CLI tool"""
import datetime
import hashlib
import os
import sys
import httplib2
from oslo_serialization import jsonutils
from oslo_utils import units
from six.moves import http_client
# NOTE(jokke): simplified transition to py3, behaves like py2 xrange
from six.moves import range
from glance.tests import functional
from glance.tests.utils import execute
from glance.tests.utils import minimal_headers
FIVE_KB = 5 * units.Ki
class TestBinGlanceCacheManage(functional.FunctionalTest):
"""Functional tests for the bin/glance CLI tool"""
def setUp(self):
self.image_cache_driver = "sqlite"
super(TestBinGlanceCacheManage, self).setUp()
self.api_server.deployment_flavor = "cachemanagement"
# NOTE(sirp): This is needed in case we are running the tests under an
# environment in which OS_AUTH_STRATEGY=keystone. The test server we
# spin up won't have keystone support, so we need to switch to the
# NoAuth strategy.
os.environ['OS_AUTH_STRATEGY'] = 'noauth'
os.environ['OS_AUTH_URL'] = ''
def add_image(self, name):
"""
Adds an image with supplied name and returns the newly-created
image identifier.
"""
image_data = b"*" * FIVE_KB
headers = minimal_headers(name)
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual(name, data['image']['name'])
self.assertTrue(data['image']['is_public'])
return data['image']['id']
def is_image_cached(self, image_id):
"""
Return True if supplied image ID is cached, False otherwise
"""
exe_cmd = '%s -m glance.cmd.cache_manage' % sys.executable
cmd = "%s --port=%d list-cached" % (exe_cmd, self.api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
out = out.decode('utf-8')
return image_id in out
def iso_date(self, image_id):
"""
Return True if supplied image ID is cached, False otherwise
"""
exe_cmd = '%s -m glance.cmd.cache_manage' % sys.executable
cmd = "%s --port=%d list-cached" % (exe_cmd, self.api_port)
exitcode, out, err = execute(cmd)
out = out.decode('utf-8')
return datetime.datetime.utcnow().strftime("%Y-%m-%d") in out
def test_no_cache_enabled(self):
"""
Test that cache index command works
"""
self.cleanup()
self.api_server.deployment_flavor = ''
self.start_servers() # Not passing in cache_manage in pipeline...
api_port = self.api_port
# Verify decent error message returned
exe_cmd = '%s -m glance.cmd.cache_manage' % sys.executable
cmd = "%s --port=%d list-cached" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd, raise_error=False)
self.assertEqual(1, exitcode)
self.assertIn(b'Cache management middleware not enabled on host',
out.strip())
self.stop_servers()
def test_cache_index(self):
"""
Test that cache index command works
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
api_port = self.api_port
# Verify no cached images
exe_cmd = '%s -m glance.cmd.cache_manage' % sys.executable
cmd = "%s --port=%d list-cached" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'No cached images', out.strip())
ids = {}
# Add a few images and cache the second one of them
# by GETing the image...
for x in range(4):
ids[x] = self.add_image("Image%s" % x)
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", api_port,
ids[1])
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertTrue(self.is_image_cached(ids[1]),
"%s is not cached." % ids[1])
self.assertTrue(self.iso_date(ids[1]))
self.stop_servers()
def test_queue(self):
"""
Test that we can queue and fetch images using the
CLI utility
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
api_port = self.api_port
# Verify no cached images
exe_cmd = '%s -m glance.cmd.cache_manage' % sys.executable
cmd = "%s --port=%d list-cached" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'No cached images', out.strip())
# Verify no queued images
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'No queued images', out.strip())
ids = {}
# Add a few images and cache the second one of them
# by GETing the image...
for x in range(4):
ids[x] = self.add_image("Image%s" % x)
# Queue second image and then cache it
cmd = "%s --port=%d --force queue-image %s" % (
exe_cmd, api_port, ids[1])
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
# Verify queued second image
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
out = out.decode('utf-8')
self.assertIn(ids[1], out, 'Image %s was not queued!' % ids[1])
# Cache images in the queue by running the prefetcher
cache_config_filepath = os.path.join(self.test_dir, 'etc',
'glance-cache.conf')
cache_file_options = {
'image_cache_dir': self.api_server.image_cache_dir,
'image_cache_driver': self.image_cache_driver,
'registry_port': self.registry_server.bind_port,
'lock_path': self.test_dir,
'log_file': os.path.join(self.test_dir, 'cache.log'),
'metadata_encryption_key': "012345678901234567890123456789ab",
'filesystem_store_datadir': self.test_dir
}
with open(cache_config_filepath, 'w') as cache_file:
cache_file.write("""[DEFAULT]
debug = True
lock_path = %(lock_path)s
image_cache_dir = %(image_cache_dir)s
image_cache_driver = %(image_cache_driver)s
registry_host = 127.0.0.1
registry_port = %(registry_port)s
metadata_encryption_key = %(metadata_encryption_key)s
log_file = %(log_file)s
[glance_store]
filesystem_store_datadir=%(filesystem_store_datadir)s
""" % cache_file_options)
cmd = ("%s -m glance.cmd.cache_prefetcher --config-file %s" %
(sys.executable, cache_config_filepath))
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual(b'', out.strip(), out)
# Verify no queued images
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'No queued images', out.strip())
# Verify second image now cached
cmd = "%s --port=%d list-cached" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
out = out.decode('utf-8')
self.assertIn(ids[1], out, 'Image %s was not cached!' % ids[1])
# Queue third image and then delete it from queue
cmd = "%s --port=%d --force queue-image %s" % (
exe_cmd, api_port, ids[2])
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
# Verify queued third image
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
out = out.decode('utf-8')
self.assertIn(ids[2], out, 'Image %s was not queued!' % ids[2])
# Delete the image from the queue
cmd = ("%s --port=%d --force "
"delete-queued-image %s") % (exe_cmd, api_port, ids[2])
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
# Verify no queued images
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'No queued images', out.strip())
# Queue all images
for x in range(4):
cmd = ("%s --port=%d --force "
"queue-image %s") % (exe_cmd, api_port, ids[x])
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
# Verify queued third image
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'Found 3 queued images', out)
# Delete the image from the queue
cmd = ("%s --port=%d --force "
"delete-all-queued-images") % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
# Verify nothing in queue anymore
cmd = "%s --port=%d list-queued" % (exe_cmd, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertIn(b'No queued images', out.strip())
# verify two image id when queue-image
cmd = ("%s --port=%d --force "
"queue-image %s %s") % (exe_cmd, api_port, ids[0], ids[1])
exitcode, out, err = execute(cmd, raise_error=False)
self.assertEqual(1, exitcode)
self.assertIn(b'Please specify one and only ID of '
b'the image you wish to ', out.strip())
# verify two image id when delete-queued-image
cmd = ("%s --port=%d --force delete-queued-image "
"%s %s") % (exe_cmd, api_port, ids[0], ids[1])
exitcode, out, err = execute(cmd, raise_error=False)
self.assertEqual(1, exitcode)
self.assertIn(b'Please specify one and only ID of '
b'the image you wish to ', out.strip())
# verify two image id when delete-cached-image
cmd = ("%s --port=%d --force delete-cached-image "
"%s %s") % (exe_cmd, api_port, ids[0], ids[1])
exitcode, out, err = execute(cmd, raise_error=False)
self.assertEqual(1, exitcode)
self.assertIn(b'Please specify one and only ID of '
b'the image you wish to ', out.strip())
self.stop_servers()

View File

@ -20,25 +20,16 @@ but that is really not relevant, as the image cache is transparent
to the backend store.
"""
import hashlib
import os
import shutil
import sys
import time
import uuid
import httplib2
from oslo_serialization import jsonutils
from oslo_utils import units
from six.moves import http_client
# NOTE(jokke): simplified transition to py3, behaves like py2 xrange
from six.moves import range
from glance.tests import functional
from glance.tests.functional.store_utils import get_http_uri
from glance.tests.functional.store_utils import setup_http
from glance.tests.utils import execute
from glance.tests.utils import minimal_headers
from glance.tests.utils import skip_if_disabled
from glance.tests.utils import xattr_writes_supported
@ -47,78 +38,6 @@ FIVE_KB = 5 * units.Ki
class BaseCacheMiddlewareTest(object):
@skip_if_disabled
def test_cache_middleware_transparent_v1(self):
"""
We test that putting the cache middleware into the
application pipeline gives us transparent image caching
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# Add an image and verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
image_id = data['image']['id']
# Verify image not in cache
image_cached_path = os.path.join(self.api_server.image_cache_dir,
image_id)
self.assertFalse(os.path.exists(image_cached_path))
# Grab the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Verify image now in cache
image_cached_path = os.path.join(self.api_server.image_cache_dir,
image_id)
# You might wonder why the heck this is here... well, it's here
# because it took me forever to figure out that the disk write
# cache in Linux was causing random failures of the os.path.exists
# assert directly below this. Basically, since the cache is writing
# the image file to disk in a different process, the write buffers
# don't flush the cache file during an os.rename() properly, resulting
# in a false negative on the file existence check below. This little
# loop pauses the execution of this process for no more than 1.5
# seconds. If after that time the cached image file still doesn't
# appear on disk, something really is wrong, and the assert should
# trigger...
i = 0
while not os.path.exists(image_cached_path) and i < 30:
time.sleep(0.05)
i = i + 1
self.assertTrue(os.path.exists(image_cached_path))
# Now, we delete the image from the server and verify that
# the image cache no longer contains the deleted image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
self.assertFalse(os.path.exists(image_cached_path))
self.stop_servers()
@skip_if_disabled
def test_cache_middleware_transparent_v2(self):
"""Ensure the v2 API image transfer calls trigger caching"""
@ -354,102 +273,6 @@ class BaseCacheMiddlewareTest(object):
self.stop_servers()
@skip_if_disabled
def test_cache_remote_image(self):
"""
We test that caching is no longer broken for remote images
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
setup_http(self)
# Add a remote image and verify a 201 Created is returned
remote_uri = get_http_uri(self, '2')
headers = {'X-Image-Meta-Name': 'Image2',
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True',
'X-Image-Meta-Location': remote_uri}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(FIVE_KB, data['image']['size'])
image_id = data['image']['id']
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
# Grab the image
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Grab the image again to ensure it can be served out from
# cache with the correct size
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual(FIVE_KB, int(response['content-length']))
self.stop_servers()
@skip_if_disabled
def test_cache_middleware_trans_v1_without_download_image_policy(self):
"""
Ensure the image v1 API image transfer applied 'download_image'
policy enforcement.
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# Add an image and verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
image_id = data['image']['id']
# Verify image not in cache
image_cached_path = os.path.join(self.api_server.image_cache_dir,
image_id)
self.assertFalse(os.path.exists(image_cached_path))
rules = {"context_is_admin": "role:admin", "default": "",
"download_image": "!"}
self.set_policy_rules(rules)
# Grab the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.FORBIDDEN, response.status)
# Now, we delete the image from the server and verify that
# the image cache no longer contains the deleted image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
self.assertFalse(os.path.exists(image_cached_path))
self.stop_servers()
@skip_if_disabled
def test_cache_middleware_trans_v2_without_download_image_policy(self):
"""
@ -511,489 +334,6 @@ class BaseCacheMiddlewareTest(object):
self.stop_servers()
@skip_if_disabled
def test_cache_middleware_trans_with_deactivated_image(self):
"""
Ensure the image v1/v2 API image transfer forbids downloading
deactivated images.
Image deactivation is not available in v1. So, we'll deactivate the
image using v2 but test image transfer with both v1 and v2.
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# Add an image and verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
image_id = data['image']['id']
# Grab the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Verify image in cache
image_cached_path = os.path.join(self.api_server.image_cache_dir,
image_id)
self.assertTrue(os.path.exists(image_cached_path))
# Deactivate the image using v2
path = "http://%s:%d/v2/images/%s/actions/deactivate"
path = path % ("127.0.0.1", self.api_port, image_id)
http = httplib2.Http()
response, content = http.request(path, 'POST')
self.assertEqual(http_client.NO_CONTENT, response.status)
# Download the image with v1. Ensure it is forbidden
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.FORBIDDEN, response.status)
# Download the image with v2. This succeeds because
# we are in admin context.
path = "http://%s:%d/v2/images/%s/file" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Reactivate the image using v2
path = "http://%s:%d/v2/images/%s/actions/reactivate"
path = path % ("127.0.0.1", self.api_port, image_id)
http = httplib2.Http()
response, content = http.request(path, 'POST')
self.assertEqual(http_client.NO_CONTENT, response.status)
# Download the image with v1. Ensure it is allowed
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Download the image with v2. Ensure it is allowed
path = "http://%s:%d/v2/images/%s/file" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Now, we delete the image from the server and verify that
# the image cache no longer contains the deleted image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
self.assertFalse(os.path.exists(image_cached_path))
self.stop_servers()
class BaseCacheManageMiddlewareTest(object):
"""Base test class for testing cache management middleware"""
def verify_no_images(self):
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('images', data)
self.assertEqual(0, len(data['images']))
def add_image(self, name):
"""
Adds an image and returns the newly-added image
identifier
"""
image_data = b"*" * FIVE_KB
headers = minimal_headers('%s' % name)
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual(name, data['image']['name'])
self.assertTrue(data['image']['is_public'])
return data['image']['id']
def verify_no_cached_images(self):
"""
Verify no images in the image cache
"""
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
self.assertEqual([], data['cached_images'])
@skip_if_disabled
def test_user_not_authorized(self):
self.cleanup()
self.start_servers(**self.__dict__.copy())
self.verify_no_images()
image_id1 = self.add_image("Image1")
image_id2 = self.add_image("Image2")
# Verify image does not yet show up in cache (we haven't "hit"
# it yet using a GET /images/1 ...
self.verify_no_cached_images()
# Grab the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id1)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Verify image now in cache
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
cached_images = data['cached_images']
self.assertEqual(1, len(cached_images))
self.assertEqual(image_id1, cached_images[0]['image_id'])
# Set policy to disallow access to cache management
rules = {"manage_image_cache": '!'}
self.set_policy_rules(rules)
# Verify an unprivileged user cannot see cached images
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.FORBIDDEN, response.status)
# Verify an unprivileged user cannot delete images from the cache
path = "http://%s:%d/v1/cached_images/%s" % ("127.0.0.1",
self.api_port, image_id1)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.FORBIDDEN, response.status)
# Verify an unprivileged user cannot delete all cached images
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.FORBIDDEN, response.status)
# Verify an unprivileged user cannot queue an image
path = "http://%s:%d/v1/queued_images/%s" % ("127.0.0.1",
self.api_port, image_id2)
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.FORBIDDEN, response.status)
self.stop_servers()
@skip_if_disabled
def test_cache_manage_get_cached_images(self):
"""
Tests that cached images are queryable
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
self.verify_no_images()
image_id = self.add_image("Image1")
# Verify image does not yet show up in cache (we haven't "hit"
# it yet using a GET /images/1 ...
self.verify_no_cached_images()
# Grab the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Verify image now in cache
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
# Verify the last_modified/last_accessed values are valid floats
for cached_image in data['cached_images']:
for time_key in ('last_modified', 'last_accessed'):
time_val = cached_image[time_key]
try:
float(time_val)
except ValueError:
self.fail('%s time %s for cached image %s not a valid '
'float' % (time_key, time_val,
cached_image['image_id']))
cached_images = data['cached_images']
self.assertEqual(1, len(cached_images))
self.assertEqual(image_id, cached_images[0]['image_id'])
self.assertEqual(0, cached_images[0]['hits'])
# Hit the image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
# Verify image hits increased in output of manage GET
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
cached_images = data['cached_images']
self.assertEqual(1, len(cached_images))
self.assertEqual(image_id, cached_images[0]['image_id'])
self.assertEqual(1, cached_images[0]['hits'])
self.stop_servers()
@skip_if_disabled
def test_cache_manage_delete_cached_images(self):
"""
Tests that cached images may be deleted
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
self.verify_no_images()
ids = {}
# Add a bunch of images...
for x in range(4):
ids[x] = self.add_image("Image%s" % str(x))
# Verify no images in cached_images because no image has been hit
# yet using a GET /images/<IMAGE_ID> ...
self.verify_no_cached_images()
# Grab the images, essentially caching them...
for x in range(4):
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
ids[x])
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status,
"Failed to find image %s" % ids[x])
# Verify images now in cache
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
cached_images = data['cached_images']
self.assertEqual(4, len(cached_images))
for x in range(4, 0): # Cached images returned last modified order
self.assertEqual(ids[x], cached_images[x]['image_id'])
self.assertEqual(0, cached_images[x]['hits'])
# Delete third image of the cached images and verify no longer in cache
path = "http://%s:%d/v1/cached_images/%s" % ("127.0.0.1",
self.api_port, ids[2])
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
cached_images = data['cached_images']
self.assertEqual(3, len(cached_images))
self.assertNotIn(ids[2], [x['image_id'] for x in cached_images])
# Delete all cached images and verify nothing in cache
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
cached_images = data['cached_images']
self.assertEqual(0, len(cached_images))
self.stop_servers()
@skip_if_disabled
def test_cache_manage_delete_queued_images(self):
"""
Tests that all queued images may be deleted at once
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
self.verify_no_images()
ids = {}
NUM_IMAGES = 4
# Add and then queue some images
for x in range(NUM_IMAGES):
ids[x] = self.add_image("Image%s" % str(x))
path = "http://%s:%d/v1/queued_images/%s" % ("127.0.0.1",
self.api_port, ids[x])
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.OK, response.status)
# Delete all queued images
path = "http://%s:%d/v1/queued_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
num_deleted = data['num_deleted']
self.assertEqual(NUM_IMAGES, num_deleted)
# Verify a second delete now returns num_deleted=0
path = "http://%s:%d/v1/queued_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
num_deleted = data['num_deleted']
self.assertEqual(0, num_deleted)
self.stop_servers()
@skip_if_disabled
def test_queue_and_prefetch(self):
"""
Tests that images may be queued and prefetched
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
cache_config_filepath = os.path.join(self.test_dir, 'etc',
'glance-cache.conf')
cache_file_options = {
'image_cache_dir': self.api_server.image_cache_dir,
'image_cache_driver': self.image_cache_driver,
'registry_port': self.registry_server.bind_port,
'log_file': os.path.join(self.test_dir, 'cache.log'),
'lock_path': self.test_dir,
'metadata_encryption_key': "012345678901234567890123456789ab",
'filesystem_store_datadir': self.test_dir
}
with open(cache_config_filepath, 'w') as cache_file:
cache_file.write("""[DEFAULT]
debug = True
lock_path = %(lock_path)s
image_cache_dir = %(image_cache_dir)s
image_cache_driver = %(image_cache_driver)s
registry_host = 127.0.0.1
registry_port = %(registry_port)s
metadata_encryption_key = %(metadata_encryption_key)s
log_file = %(log_file)s
[glance_store]
filesystem_store_datadir=%(filesystem_store_datadir)s
""" % cache_file_options)
self.verify_no_images()
ids = {}
# Add a bunch of images...
for x in range(4):
ids[x] = self.add_image("Image%s" % str(x))
# Queue the first image, verify no images still in cache after queueing
# then run the prefetcher and verify that the image is then in the
# cache
path = "http://%s:%d/v1/queued_images/%s" % ("127.0.0.1",
self.api_port, ids[0])
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.OK, response.status)
self.verify_no_cached_images()
cmd = ("%s -m glance.cmd.cache_prefetcher --config-file %s" %
(sys.executable, cache_config_filepath))
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual(b'', out.strip(), out)
# Verify first image now in cache
path = "http://%s:%d/v1/cached_images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertIn('cached_images', data)
cached_images = data['cached_images']
self.assertEqual(1, len(cached_images))
self.assertIn(ids[0], [r['image_id']
for r in data['cached_images']])
self.stop_servers()
class TestImageCacheXattr(functional.FunctionalTest,
BaseCacheMiddlewareTest):
@ -1038,52 +378,6 @@ class TestImageCacheXattr(functional.FunctionalTest,
shutil.rmtree(self.api_server.image_cache_dir)
class TestImageCacheManageXattr(functional.FunctionalTest,
BaseCacheManageMiddlewareTest):
"""
Functional tests that exercise the image cache management
with the Xattr cache driver
"""
def setUp(self):
"""
Test to see if the pre-requisites for the image cache
are working (python-xattr installed and xattr support on the
filesystem)
"""
if getattr(self, 'disabled', False):
return
if not getattr(self, 'inited', False):
try:
import xattr # noqa
except ImportError:
self.inited = True
self.disabled = True
self.disabled_message = ("python-xattr not installed.")
return
self.inited = True
self.disabled = False
self.image_cache_driver = "xattr"
super(TestImageCacheManageXattr, self).setUp()
self.api_server.deployment_flavor = "cachemanagement"
if not xattr_writes_supported(self.test_dir):
self.inited = True
self.disabled = True
self.disabled_message = ("filesystem does not support xattr")
return
def tearDown(self):
super(TestImageCacheManageXattr, self).tearDown()
if os.path.exists(self.api_server.image_cache_dir):
shutil.rmtree(self.api_server.image_cache_dir)
class TestImageCacheSqlite(functional.FunctionalTest,
BaseCacheMiddlewareTest):
@ -1121,43 +415,3 @@ class TestImageCacheSqlite(functional.FunctionalTest,
super(TestImageCacheSqlite, self).tearDown()
if os.path.exists(self.api_server.image_cache_dir):
shutil.rmtree(self.api_server.image_cache_dir)
class TestImageCacheManageSqlite(functional.FunctionalTest,
BaseCacheManageMiddlewareTest):
"""
Functional tests that exercise the image cache management using the
SQLite driver
"""
def setUp(self):
"""
Test to see if the pre-requisites for the image cache
are working (python-xattr installed and xattr support on the
filesystem)
"""
if getattr(self, 'disabled', False):
return
if not getattr(self, 'inited', False):
try:
import sqlite3 # noqa
except ImportError:
self.inited = True
self.disabled = True
self.disabled_message = ("python-sqlite3 not installed.")
return
self.inited = True
self.disabled = False
self.image_cache_driver = "sqlite"
super(TestImageCacheManageSqlite, self).setUp()
self.api_server.deployment_flavor = "cachemanagement"
def tearDown(self):
super(TestImageCacheManageSqlite, self).tearDown()
if os.path.exists(self.api_server.image_cache_dir):
shutil.rmtree(self.api_server.image_cache_dir)

View File

@ -33,7 +33,7 @@ class TestCORSMiddleware(functional.FunctionalTest):
# Cleanup is handled in teardown of the parent class.
self.start_servers(**self.__dict__.copy())
self.http = httplib2.Http()
self.api_path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
self.api_path = "http://%s:%d/v2/images" % ("127.0.0.1", self.api_port)
def test_valid_cors_options_request(self):
(r_headers, content) = self.http.request(

View File

@ -1,33 +0,0 @@
# 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.
"""Functional test cases for glance-replicator"""
import sys
from glance.tests import functional
from glance.tests.utils import execute
class TestGlanceReplicator(functional.FunctionalTest):
"""Functional tests for glance-replicator"""
def test_compare(self):
# Test for issue: https://bugs.launchpad.net/glance/+bug/1598928
cmd = ('%s -m glance.cmd.replicator '
'compare az1:9292 az2:9292 --debug' %
(sys.executable,))
exitcode, out, err = execute(cmd, raise_error=False)
self.assertIn(
b'Request: GET http://az1:9292/v1/images/detail?is_public=None',
err
)

View File

@ -49,6 +49,8 @@ class TestImages(functional.FunctionalTest):
"foo_image%d" % i)
setattr(self, 'http_server%d_pid' % i, ret[0])
setattr(self, 'http_port%d' % i, ret[1])
self.api_server.use_user_token = True
self.api_server.send_identity_credentials = True
def tearDown(self):
for i in range(3):
@ -72,36 +74,6 @@ class TestImages(functional.FunctionalTest):
base_headers.update(custom_headers or {})
return base_headers
def test_v1_none_properties_v2(self):
self.api_server.deployment_flavor = 'noauth'
self.api_server.use_user_token = True
self.api_server.send_identity_credentials = True
self.registry_server.deployment_flavor = ''
# Image list should be empty
self.start_servers(**self.__dict__.copy())
# Create an image (with two deployer-defined properties)
path = self._url('/v1/images')
headers = self._headers({'content-type': 'application/octet-stream'})
headers.update(test_utils.minimal_headers('image-1'))
# NOTE(flaper87): Sending empty string, the server will use None
headers['x-image-meta-property-my_empty_prop'] = ''
response = requests.post(path, headers=headers)
self.assertEqual(http.CREATED, response.status_code)
data = jsonutils.loads(response.text)
image_id = data['image']['id']
# NOTE(flaper87): Get the image using V2 and verify
# the returned value for `my_empty_prop` is an empty
# string.
path = self._url('/v2/images/%s' % image_id)
response = requests.get(path, headers=self._headers())
self.assertEqual(http.OK, response.status_code)
image = jsonutils.loads(response.text)
self.assertEqual('', image['my_empty_prop'])
self.stop_servers()
def test_not_authenticated_in_registry_on_ops(self):
# https://bugs.launchpad.net/glance/+bug/1451850
# this configuration guarantees that authentication succeeds in
@ -3620,6 +3592,7 @@ class TestImagesWithRegistry(TestImages):
self.api_server.data_api = (
'glance.tests.functional.v2.registry_data_api')
self.registry_server.deployment_flavor = 'trusted-auth'
self.api_server.use_user_token = True
class TestImagesIPv6(functional.FunctionalTest):

View File

@ -1,222 +0,0 @@
# 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 atexit
import os.path
import tempfile
import fixtures
import glance_store
from oslo_config import cfg
from oslo_db import options
import glance.common.client
from glance.common import config
import glance.db.sqlalchemy.api
import glance.registry.client.v1.client
from glance import tests as glance_tests
from glance.tests import utils as test_utils
TESTING_API_PASTE_CONF = """
[pipeline:glance-api]
pipeline = versionnegotiation gzip unauthenticated-context rootapp
[pipeline:glance-api-caching]
pipeline = versionnegotiation gzip unauthenticated-context cache rootapp
[pipeline:glance-api-cachemanagement]
pipeline =
versionnegotiation
gzip
unauthenticated-context
cache
cache_manage
rootapp
[pipeline:glance-api-fakeauth]
pipeline = versionnegotiation gzip fakeauth context rootapp
[pipeline:glance-api-noauth]
pipeline = versionnegotiation gzip context rootapp
[composite:rootapp]
paste.composite_factory = glance.api:root_app_factory
/: apiversions
/v1: apiv1app
/v2: apiv2app
[app:apiversions]
paste.app_factory = glance.api.versions:create_resource
[app:apiv1app]
paste.app_factory = glance.api.v1.router:API.factory
[app:apiv2app]
paste.app_factory = glance.api.v2.router:API.factory
[filter:versionnegotiation]
paste.filter_factory =
glance.api.middleware.version_negotiation:VersionNegotiationFilter.factory
[filter:gzip]
paste.filter_factory = glance.api.middleware.gzip:GzipMiddleware.factory
[filter:cache]
paste.filter_factory = glance.api.middleware.cache:CacheFilter.factory
[filter:cache_manage]
paste.filter_factory =
glance.api.middleware.cache_manage:CacheManageFilter.factory
[filter:context]
paste.filter_factory = glance.api.middleware.context:ContextMiddleware.factory
[filter:unauthenticated-context]
paste.filter_factory =
glance.api.middleware.context:UnauthenticatedContextMiddleware.factory
[filter:fakeauth]
paste.filter_factory = glance.tests.utils:FakeAuthMiddleware.factory
"""
TESTING_REGISTRY_PASTE_CONF = """
[pipeline:glance-registry]
pipeline = unauthenticated-context registryapp
[pipeline:glance-registry-fakeauth]
pipeline = fakeauth context registryapp
[app:registryapp]
paste.app_factory = glance.registry.api.v1:API.factory
[filter:context]
paste.filter_factory = glance.api.middleware.context:ContextMiddleware.factory
[filter:unauthenticated-context]
paste.filter_factory =
glance.api.middleware.context:UnauthenticatedContextMiddleware.factory
[filter:fakeauth]
paste.filter_factory = glance.tests.utils:FakeAuthMiddleware.factory
"""
CONF = cfg.CONF
class ApiTest(test_utils.BaseTestCase):
def setUp(self):
super(ApiTest, self).setUp()
self.init()
def init(self):
self.test_dir = self.useFixture(fixtures.TempDir()).path
self._configure_logging()
self._configure_policy()
self._setup_database()
self._setup_stores()
self._setup_property_protection()
self.glance_registry_app = self._load_paste_app(
'glance-registry',
flavor=getattr(self, 'registry_flavor', ''),
conf=getattr(self, 'registry_paste_conf',
TESTING_REGISTRY_PASTE_CONF),
)
self._connect_registry_client()
self.glance_api_app = self._load_paste_app(
'glance-api',
flavor=getattr(self, 'api_flavor', ''),
conf=getattr(self, 'api_paste_conf', TESTING_API_PASTE_CONF),
)
self.http = test_utils.Httplib2WsgiAdapter(self.glance_api_app)
def _setup_property_protection(self):
self._copy_data_file('property-protections.conf', self.test_dir)
self.property_file = os.path.join(self.test_dir,
'property-protections.conf')
def _configure_policy(self):
policy_file = self._copy_data_file('policy.json', self.test_dir)
self.config(policy_file=policy_file, group='oslo_policy')
def _configure_logging(self):
self.config(default_log_levels=[
'amqplib=WARN',
'sqlalchemy=WARN',
'boto=WARN',
'suds=INFO',
'keystone=INFO',
'eventlet.wsgi.server=DEBUG'
])
def _setup_database(self):
sql_connection = 'sqlite:////%s/tests.sqlite' % self.test_dir
options.set_defaults(CONF, connection=sql_connection)
glance.db.sqlalchemy.api.clear_db_env()
glance_db_env = 'GLANCE_DB_TEST_SQLITE_FILE'
if glance_db_env in os.environ:
# use the empty db created and cached as a tempfile
# instead of spending the time creating a new one
db_location = os.environ[glance_db_env]
test_utils.execute('cp %s %s/tests.sqlite'
% (db_location, self.test_dir))
else:
test_utils.db_sync()
# copy the clean db to a temp location so that it
# can be reused for future tests
(osf, db_location) = tempfile.mkstemp()
os.close(osf)
test_utils.execute('cp %s/tests.sqlite %s'
% (self.test_dir, db_location))
os.environ[glance_db_env] = db_location
# cleanup the temp file when the test suite is
# complete
def _delete_cached_db():
try:
os.remove(os.environ[glance_db_env])
except Exception:
glance_tests.logger.exception(
"Error cleaning up the file %s" %
os.environ[glance_db_env])
atexit.register(_delete_cached_db)
def _setup_stores(self):
glance_store.register_opts(CONF)
image_dir = os.path.join(self.test_dir, "images")
self.config(group='glance_store',
filesystem_store_datadir=image_dir)
glance_store.create_stores()
def _load_paste_app(self, name, flavor, conf):
conf_file_path = os.path.join(self.test_dir, '%s-paste.ini' % name)
with open(conf_file_path, 'w') as conf_file:
conf_file.write(conf)
conf_file.flush()
return config.load_paste_app(name, flavor=flavor,
conf_file=conf_file_path)
def _connect_registry_client(self):
def get_connection_type(self2):
def wrapped(*args, **kwargs):
return test_utils.HttplibWsgiAdapter(self.glance_registry_app)
return wrapped
self.stubs.Set(glance.common.client.BaseClient,
'get_connection_type', get_connection_type)
def tearDown(self):
glance.db.sqlalchemy.api.clear_db_env()
super(ApiTest, self).tearDown()

File diff suppressed because it is too large Load Diff

View File

@ -1,298 +0,0 @@
# 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 argparse
import sys
import mock
import prettytable
from glance.cmd import cache_manage
from glance.common import exception
import glance.common.utils
import glance.image_cache.client
from glance.tests import utils as test_utils
@mock.patch('sys.stdout', mock.Mock())
class TestGlanceCmdManage(test_utils.BaseTestCase):
def _run_command(self, cmd_args, return_code=None):
"""Runs the cache-manage command.
:param cmd_args: The command line arguments.
:param return_code: The expected return code of the command.
"""
testargs = ['cache_manage']
testargs.extend(cmd_args)
with mock.patch.object(sys, 'exit') as mock_exit:
with mock.patch.object(sys, 'argv', testargs):
try:
cache_manage.main()
except Exception:
# See if we expected this failure
if return_code is None:
raise
if return_code is not None:
mock_exit.called_with(return_code)
@mock.patch.object(argparse.ArgumentParser, 'print_help')
def test_help(self, mock_print_help):
self._run_command(['help'])
self.assertEqual(1, mock_print_help.call_count)
@mock.patch.object(cache_manage, 'lookup_command')
def test_help_with_command(self, mock_lookup_command):
mock_lookup_command.return_value = cache_manage.print_help
self._run_command(['help', 'list-cached'])
mock_lookup_command.assert_any_call('help')
mock_lookup_command.assert_any_call('list-cached')
def test_help_with_redundant_command(self):
self._run_command(['help', 'list-cached', '1'], cache_manage.FAILURE)
@mock.patch.object(glance.image_cache.client.CacheClient,
'get_cached_images')
@mock.patch.object(prettytable.PrettyTable, 'add_row')
def test_list_cached_images(self, mock_row_create, mock_images):
"""
Verify that list_cached() method correctly processes images with all
filled data and images with not filled 'last_accessed' field.
"""
mock_images.return_value = [
{'last_accessed': float(0),
'last_modified': float(1378985797.124511),
'image_id': '1', 'size': '128', 'hits': '1'},
{'last_accessed': float(1378985797.124511),
'last_modified': float(1378985797.124511),
'image_id': '2', 'size': '255', 'hits': '2'}]
self._run_command(['list-cached'], cache_manage.SUCCESS)
self.assertEqual(len(mock_images.return_value),
mock_row_create.call_count)
@mock.patch.object(glance.image_cache.client.CacheClient,
'get_cached_images')
def test_list_cached_images_empty(self, mock_images):
"""
Verify that list_cached() method handles a case when no images are
cached without errors.
"""
self._run_command(['list-cached'], cache_manage.SUCCESS)
@mock.patch.object(glance.image_cache.client.CacheClient,
'get_queued_images')
@mock.patch.object(prettytable.PrettyTable, 'add_row')
def test_list_queued_images(self, mock_row_create, mock_images):
"""Verify that list_queued() method correctly processes images."""
mock_images.return_value = [
{'image_id': '1'}, {'image_id': '2'}]
# cache_manage.list_queued(mock.Mock())
self._run_command(['list-queued'], cache_manage.SUCCESS)
self.assertEqual(len(mock_images.return_value),
mock_row_create.call_count)
@mock.patch.object(glance.image_cache.client.CacheClient,
'get_queued_images')
def test_list_queued_images_empty(self, mock_images):
"""
Verify that list_queued() method handles a case when no images were
queued without errors.
"""
mock_images.return_value = []
self._run_command(['list-queued'], cache_manage.SUCCESS)
def test_queue_image_without_index(self):
self._run_command(['queue-image'], cache_manage.FAILURE)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_queue_image_not_forced_not_confirmed(self,
mock_client, mock_confirm):
# --force not set and queue confirmation return False.
mock_confirm.return_value = False
self._run_command(['queue-image', 'fakeimageid'], cache_manage.SUCCESS)
self.assertFalse(mock_client.called)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_queue_image_not_forced_confirmed(self, mock_get_client,
mock_confirm):
# --force not set and confirmation return True.
mock_confirm.return_value = True
mock_client = mock.MagicMock()
mock_get_client.return_value = mock_client
# verbose to cover additional condition and line
self._run_command(['queue-image', 'fakeimageid', '-v'],
cache_manage.SUCCESS)
self.assertTrue(mock_get_client.called)
mock_client.queue_image_for_caching.assert_called_with('fakeimageid')
def test_delete_cached_image_without_index(self):
self._run_command(['delete-cached-image'], cache_manage.FAILURE)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_cached_image_not_forced_not_confirmed(self,
mock_client,
mock_confirm):
# --force not set and confirmation return False.
mock_confirm.return_value = False
self._run_command(['delete-cached-image', 'fakeimageid'],
cache_manage.SUCCESS)
self.assertFalse(mock_client.called)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_cached_image_not_forced_confirmed(self, mock_get_client,
mock_confirm):
# --force not set and confirmation return True.
mock_confirm.return_value = True
mock_client = mock.MagicMock()
mock_get_client.return_value = mock_client
# verbose to cover additional condition and line
self._run_command(['delete-cached-image', 'fakeimageid', '-v'],
cache_manage.SUCCESS)
self.assertTrue(mock_get_client.called)
mock_client.delete_cached_image.assert_called_with('fakeimageid')
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_cached_images_not_forced_not_confirmed(self,
mock_client,
mock_confirm):
# --force not set and confirmation return False.
mock_confirm.return_value = False
self._run_command(['delete-all-cached-images'], cache_manage.SUCCESS)
self.assertFalse(mock_client.called)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_cached_images_not_forced_confirmed(self, mock_get_client,
mock_confirm):
# --force not set and confirmation return True.
mock_confirm.return_value = True
mock_client = mock.MagicMock()
mock_get_client.return_value = mock_client
# verbose to cover additional condition and line
self._run_command(['delete-all-cached-images', '-v'],
cache_manage.SUCCESS)
self.assertTrue(mock_get_client.called)
mock_client.delete_all_cached_images.assert_called()
def test_delete_queued_image_without_index(self):
self._run_command(['delete-queued-image'], cache_manage.FAILURE)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_queued_image_not_forced_not_confirmed(self,
mock_client,
mock_confirm):
# --force not set and confirmation set to False.
mock_confirm.return_value = False
self._run_command(['delete-queued-image', 'img_id'],
cache_manage.SUCCESS)
self.assertFalse(mock_client.called)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_queued_image_not_forced_confirmed(self, mock_get_client,
mock_confirm):
# --force not set and confirmation set to True.
mock_confirm.return_value = True
mock_client = mock.MagicMock()
mock_get_client.return_value = mock_client
self._run_command(['delete-queued-image', 'img_id', '-v'],
cache_manage.SUCCESS)
self.assertTrue(mock_get_client.called)
mock_client.delete_queued_image.assert_called_with('img_id')
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_queued_images_not_forced_not_confirmed(self,
mock_client,
mock_confirm):
# --force not set and confirmation set to False.
mock_confirm.return_value = False
self._run_command(['delete-all-queued-images'],
cache_manage.SUCCESS)
self.assertFalse(mock_client.called)
@mock.patch.object(glance.cmd.cache_manage, 'user_confirm')
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_delete_queued_images_not_forced_confirmed(self, mock_get_client,
mock_confirm):
# --force not set and confirmation set to True.
mock_confirm.return_value = True
mock_client = mock.MagicMock()
mock_get_client.return_value = mock_client
self._run_command(['delete-all-queued-images', '-v'],
cache_manage.SUCCESS)
self.assertTrue(mock_get_client.called)
mock_client.delete_all_queued_images.assert_called()
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_catch_error_not_found(self, mock_function):
mock_function.side_effect = exception.NotFound()
self.assertEqual(cache_manage.FAILURE,
cache_manage.list_cached(mock.Mock()))
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_catch_error_forbidden(self, mock_function):
mock_function.side_effect = exception.Forbidden()
self.assertEqual(cache_manage.FAILURE,
cache_manage.list_cached(mock.Mock()))
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_catch_error_unhandled(self, mock_function):
mock_function.side_effect = exception.Duplicate()
my_mock = mock.Mock()
my_mock.debug = False
self.assertEqual(cache_manage.FAILURE,
cache_manage.list_cached(my_mock))
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
def test_catch_error_unhandled_debug_mode(self, mock_function):
mock_function.side_effect = exception.Duplicate()
my_mock = mock.Mock()
my_mock.debug = True
self.assertRaises(exception.Duplicate,
cache_manage.list_cached, my_mock)
def test_cache_manage_env(self):
def_value = 'sometext12345678900987654321'
self.assertNotEqual(def_value,
cache_manage.env('PATH', default=def_value))
def test_cache_manage_env_default(self):
def_value = 'sometext12345678900987654321'
self.assertEqual(def_value,
cache_manage.env('TMPVALUE1234567890',
default=def_value))
def test_lookup_command_unsupported_command(self):
self._run_command(['unsupported_command'], cache_manage.FAILURE)

View File

@ -17,9 +17,7 @@ import testtools
import webob
import glance.api.common
from glance.common import config
from glance.common import exception
from glance.tests import utils as test_utils
class SimpleIterator(object):
@ -126,20 +124,3 @@ class TestSizeCheckedIter(testtools.TestCase):
self.assertEqual('CD', next(checked_image))
self.assertEqual('E', next(checked_image))
self.assertRaises(exception.GlanceException, next, checked_image)
class TestMalformedRequest(test_utils.BaseTestCase):
def setUp(self):
"""Establish a clean test environment"""
super(TestMalformedRequest, self).setUp()
self.config(flavor='',
group='paste_deploy',
config_file='etc/glance-api-paste.ini')
self.api = config.load_paste_app('glance-api')
def test_redirect_incomplete_url(self):
"""Test Glance redirects /v# to /v#/ with correct Location header"""
req = webob.Request.blank('/v1.1')
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPFound.code, res.status_int)
self.assertEqual('http://localhost/v1/', res.location)

View File

@ -31,7 +31,6 @@ import six
from six.moves import http_client as http
import webob
from glance.api.v1 import router as router_v1
from glance.api.v2 import router as router_v2
from glance.common import exception
from glance.common import utils
@ -219,24 +218,6 @@ class RequestTest(test_utils.BaseTestCase):
def test_http_error_response_codes(self):
sample_id, member_id, tag_val, task_id = 'abc', '123', '1', '2'
"""Makes sure v1 unallowed methods return 405"""
unallowed_methods = [
('/images', ['PUT', 'DELETE', 'HEAD', 'PATCH']),
('/images/detail', ['POST', 'PUT', 'DELETE', 'PATCH']),
('/images/%s' % sample_id, ['POST', 'PATCH']),
('/images/%s/members' % sample_id,
['POST', 'DELETE', 'HEAD', 'PATCH']),
('/images/%s/members/%s' % (sample_id, member_id),
['POST', 'HEAD', 'PATCH']),
]
api = test_utils.FakeAuthMiddleware(router_v1.API(routes.Mapper()))
for uri, methods in unallowed_methods:
for method in methods:
req = webob.Request.blank(uri)
req.method = method
res = req.get_response(api)
self.assertEqual(http.METHOD_NOT_ALLOWED, res.status_int)
"""Makes sure v2 unallowed methods return 405"""
unallowed_methods = [
('/schemas/image', ['POST', 'PUT', 'DELETE', 'PATCH', 'HEAD']),

View File

@ -24,7 +24,6 @@ import glance.api.middleware.cache
import glance.api.policy
from glance.common import exception
from glance import context
import glance.registry.client.v1.api as registry
from glance.tests.unit import base
from glance.tests.unit import utils as unit_test_utils
@ -42,21 +41,6 @@ class ImageStub(object):
class TestCacheMiddlewareURLMatching(testtools.TestCase):
def test_v1_no_match_detail(self):
req = webob.Request.blank('/v1/images/detail')
out = glance.api.middleware.cache.CacheFilter._match_request(req)
self.assertIsNone(out)
def test_v1_no_match_detail_with_query_params(self):
req = webob.Request.blank('/v1/images/detail?limit=10')
out = glance.api.middleware.cache.CacheFilter._match_request(req)
self.assertIsNone(out)
def test_v1_match_id_with_query_param(self):
req = webob.Request.blank('/v1/images/asdf?ping=pong')
out = glance.api.middleware.cache.CacheFilter._match_request(req)
self.assertEqual(('v1', 'GET', 'asdf'), out)
def test_v2_match_id(self):
req = webob.Request.blank('/v2/images/asdf/file')
out = glance.api.middleware.cache.CacheFilter._match_request(req)
@ -180,139 +164,6 @@ class TestCacheMiddlewareProcessRequest(base.IsolatedUnitTest):
enforcer.set_rules(rules, overwrite=True)
return enforcer
def test_v1_deleted_image_fetch(self):
"""
Test for determining that when an admin tries to download a deleted
image it returns 404 Not Found error.
"""
def dummy_img_iterator():
for i in range(3):
yield i
image_id = 'test1'
image_meta = {
'id': image_id,
'name': 'fake_image',
'status': 'deleted',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': True,
'updated_at': '',
'properties': {},
}
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
cache_filter = ProcessRequestTestCacheFilter()
self.assertRaises(exception.NotFound, cache_filter._process_v1_request,
request, image_id, dummy_img_iterator, image_meta)
def test_process_v1_request_for_deleted_but_cached_image(self):
"""
Test for determining image is deleted from cache when it is not found
in Glance Registry.
"""
def fake_process_v1_request(request, image_id, image_iterator,
image_meta):
raise exception.ImageNotFound()
def fake_get_v1_image_metadata(request, image_id):
return {'status': 'active', 'properties': {}}
image_id = 'test1'
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
cache_filter = ProcessRequestTestCacheFilter()
self.stubs.Set(cache_filter, '_get_v1_image_metadata',
fake_get_v1_image_metadata)
self.stubs.Set(cache_filter, '_process_v1_request',
fake_process_v1_request)
cache_filter.process_request(request)
self.assertIn(image_id, cache_filter.cache.deleted_images)
def test_v1_process_request_image_fetch(self):
def dummy_img_iterator():
for i in range(3):
yield i
image_id = 'test1'
image_meta = {
'id': image_id,
'name': 'fake_image',
'status': 'active',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': False,
'updated_at': '',
'properties': {},
}
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
cache_filter = ProcessRequestTestCacheFilter()
actual = cache_filter._process_v1_request(
request, image_id, dummy_img_iterator, image_meta)
self.assertTrue(actual)
def test_v1_remove_location_image_fetch(self):
class CheckNoLocationDataSerializer(object):
def show(self, response, raw_response):
return 'location_data' in raw_response['image_meta']
def dummy_img_iterator():
for i in range(3):
yield i
image_id = 'test1'
image_meta = {
'id': image_id,
'name': 'fake_image',
'status': 'active',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': False,
'updated_at': '',
'properties': {},
}
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
cache_filter = ProcessRequestTestCacheFilter()
cache_filter.serializer = CheckNoLocationDataSerializer()
actual = cache_filter._process_v1_request(
request, image_id, dummy_img_iterator, image_meta)
self.assertFalse(actual)
def test_verify_metadata_deleted_image(self):
"""
Test verify_metadata raises exception.NotFound for a deleted image
@ -439,119 +290,6 @@ class TestCacheMiddlewareProcessRequest(base.IsolatedUnitTest):
self.assertRaises(webob.exc.HTTPForbidden,
cache_filter.process_request, request)
def test_v1_process_request_download_restricted(self):
"""
Test process_request for v1 api where _member_ role not able to
download the image with custom property.
"""
image_id = 'test1'
def fake_get_v1_image_metadata(*args, **kwargs):
return {
'id': image_id,
'name': 'fake_image',
'status': 'active',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': False,
'updated_at': '',
'x_test_key': 'test_1234'
}
enforcer = self._enforcer_from_rules({
"restricted":
"not ('test_1234':%(x_test_key)s and role:_member_)",
"download_image": "role:admin or rule:restricted"
})
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext(roles=['_member_'])
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._get_v1_image_metadata = fake_get_v1_image_metadata
cache_filter.policy = enforcer
self.assertRaises(webob.exc.HTTPForbidden,
cache_filter.process_request, request)
def test_v1_process_request_download_permitted(self):
"""
Test process_request for v1 api where member role able to
download the image with custom property.
"""
image_id = 'test1'
def fake_get_v1_image_metadata(*args, **kwargs):
return {
'id': image_id,
'name': 'fake_image',
'status': 'active',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': False,
'updated_at': '',
'x_test_key': 'test_1234'
}
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext(roles=['member'])
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._get_v1_image_metadata = fake_get_v1_image_metadata
rules = {
"restricted":
"not ('test_1234':%(x_test_key)s and role:_member_)",
"download_image": "role:admin or rule:restricted"
}
self.set_policy_rules(rules)
cache_filter.policy = glance.api.policy.Enforcer()
actual = cache_filter.process_request(request)
self.assertTrue(actual)
def test_v1_process_request_image_meta_not_found(self):
"""
Test process_request for v1 api where registry raises NotFound
exception as image metadata not found.
"""
image_id = 'test1'
def fake_get_v1_image_metadata(*args, **kwargs):
raise exception.NotFound()
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext(roles=['_member_'])
cache_filter = ProcessRequestTestCacheFilter()
self.stubs.Set(registry, 'get_image_metadata',
fake_get_v1_image_metadata)
rules = {
"restricted":
"not ('test_1234':%(x_test_key)s and role:_member_)",
"download_image": "role:admin or rule:restricted"
}
self.set_policy_rules(rules)
cache_filter.policy = glance.api.policy.Enforcer()
self.assertRaises(webob.exc.HTTPNotFound,
cache_filter.process_request, request)
def test_v2_process_request_download_restricted(self):
"""
Test process_request for v2 api where _member_ role not able to
@ -614,15 +352,6 @@ class TestCacheMiddlewareProcessRequest(base.IsolatedUnitTest):
class TestCacheMiddlewareProcessResponse(base.IsolatedUnitTest):
def test_process_v1_DELETE_response(self):
image_id = 'test1'
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
cache_filter = ProcessRequestTestCacheFilter()
headers = {"x-image-meta-deleted": True}
resp = webob.Response(request=request, headers=headers)
actual = cache_filter._process_DELETE_response(resp, image_id)
self.assertEqual(resp, actual)
def test_get_status_code(self):
headers = {"x-image-meta-deleted": True}
@ -631,181 +360,6 @@ class TestCacheMiddlewareProcessResponse(base.IsolatedUnitTest):
actual = cache_filter.get_status_code(resp)
self.assertEqual(http.OK, actual)
def test_process_response(self):
def fake_fetch_request_info(*args, **kwargs):
return ('test1', 'GET', 'v1')
def fake_get_v1_image_metadata(*args, **kwargs):
return {'properties': {}}
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._fetch_request_info = fake_fetch_request_info
cache_filter._get_v1_image_metadata = fake_get_v1_image_metadata
image_id = 'test1'
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
headers = {"x-image-meta-deleted": True}
resp = webob.Response(request=request, headers=headers)
actual = cache_filter.process_response(resp)
self.assertEqual(resp, actual)
def test_process_response_without_download_image_policy(self):
"""
Test for cache middleware raise webob.exc.HTTPForbidden directly
when request context has not 'download_image' role.
"""
def fake_fetch_request_info(*args, **kwargs):
return ('test1', 'GET', 'v1')
def fake_get_v1_image_metadata(*args, **kwargs):
return {'properties': {}}
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._fetch_request_info = fake_fetch_request_info
cache_filter._get_v1_image_metadata = fake_get_v1_image_metadata
rules = {'download_image': '!'}
self.set_policy_rules(rules)
cache_filter.policy = glance.api.policy.Enforcer()
image_id = 'test1'
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext()
resp = webob.Response(request=request)
self.assertRaises(webob.exc.HTTPForbidden,
cache_filter.process_response, resp)
self.assertEqual([b''], resp.app_iter)
def test_v1_process_response_download_restricted(self):
"""
Test process_response for v1 api where _member_ role not able to
download the image with custom property.
"""
image_id = 'test1'
def fake_fetch_request_info(*args, **kwargs):
return ('test1', 'GET', 'v1')
def fake_get_v1_image_metadata(*args, **kwargs):
return {
'id': image_id,
'name': 'fake_image',
'status': 'active',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': False,
'updated_at': '',
'x_test_key': 'test_1234'
}
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._fetch_request_info = fake_fetch_request_info
cache_filter._get_v1_image_metadata = fake_get_v1_image_metadata
rules = {
"restricted":
"not ('test_1234':%(x_test_key)s and role:_member_)",
"download_image": "role:admin or rule:restricted"
}
self.set_policy_rules(rules)
cache_filter.policy = glance.api.policy.Enforcer()
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext(roles=['_member_'])
resp = webob.Response(request=request)
self.assertRaises(webob.exc.HTTPForbidden,
cache_filter.process_response, resp)
def test_v1_process_response_download_permitted(self):
"""
Test process_response for v1 api where member role able to
download the image with custom property.
"""
image_id = 'test1'
def fake_fetch_request_info(*args, **kwargs):
return ('test1', 'GET', 'v1')
def fake_get_v1_image_metadata(*args, **kwargs):
return {
'id': image_id,
'name': 'fake_image',
'status': 'active',
'created_at': '',
'min_disk': '10G',
'min_ram': '1024M',
'protected': False,
'locations': '',
'checksum': 'c1234',
'owner': '',
'disk_format': 'raw',
'container_format': 'bare',
'size': '123456789',
'virtual_size': '123456789',
'is_public': 'public',
'deleted': False,
'updated_at': '',
'x_test_key': 'test_1234'
}
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._fetch_request_info = fake_fetch_request_info
cache_filter._get_v1_image_metadata = fake_get_v1_image_metadata
rules = {
"restricted":
"not ('test_1234':%(x_test_key)s and role:_member_)",
"download_image": "role:admin or rule:restricted"
}
self.set_policy_rules(rules)
cache_filter.policy = glance.api.policy.Enforcer()
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext(roles=['member'])
resp = webob.Response(request=request)
actual = cache_filter.process_response(resp)
self.assertEqual(resp, actual)
def test_v1_process_response_image_meta_not_found(self):
"""
Test process_response for v1 api where registry raises NotFound
exception as image metadata not found.
"""
image_id = 'test1'
def fake_fetch_request_info(*args, **kwargs):
return ('test1', 'GET', 'v1')
def fake_get_v1_image_metadata(*args, **kwargs):
raise exception.NotFound()
cache_filter = ProcessRequestTestCacheFilter()
cache_filter._fetch_request_info = fake_fetch_request_info
self.stubs.Set(registry, 'get_image_metadata',
fake_get_v1_image_metadata)
rules = {
"restricted":
"not ('test_1234':%(x_test_key)s and role:_member_)",
"download_image": "role:admin or rule:restricted"
}
self.set_policy_rules(rules)
cache_filter.policy = glance.api.policy.Enforcer()
request = webob.Request.blank('/v1/images/%s' % image_id)
request.context = context.RequestContext(roles=['_member_'])
resp = webob.Response(request=request)
self.assertRaises(webob.exc.HTTPNotFound,
cache_filter.process_response, resp)
def test_v2_process_response_download_restricted(self):
"""
Test process_response for v2 api where _member_ role not able to

View File

@ -1,132 +0,0 @@
# Copyright 2013 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 os
import mock
from glance.common import exception
from glance.image_cache import client
from glance.tests import utils
class CacheClientTestCase(utils.BaseTestCase):
def setUp(self):
super(CacheClientTestCase, self).setUp()
self.client = client.CacheClient('test_host')
self.client.do_request = mock.Mock()
def test_delete_cached_image(self):
self.client.do_request.return_value = utils.FakeHTTPResponse()
self.assertTrue(self.client.delete_cached_image('test_id'))
self.client.do_request.assert_called_with("DELETE",
"/cached_images/test_id")
def test_get_cached_images(self):
expected_data = b'{"cached_images": "some_images"}'
self.client.do_request.return_value = utils.FakeHTTPResponse(
data=expected_data)
self.assertEqual("some_images", self.client.get_cached_images())
self.client.do_request.assert_called_with("GET", "/cached_images")
def test_get_queued_images(self):
expected_data = b'{"queued_images": "some_images"}'
self.client.do_request.return_value = utils.FakeHTTPResponse(
data=expected_data)
self.assertEqual("some_images", self.client.get_queued_images())
self.client.do_request.assert_called_with("GET", "/queued_images")
def test_delete_all_cached_images(self):
expected_data = b'{"num_deleted": 4}'
self.client.do_request.return_value = utils.FakeHTTPResponse(
data=expected_data)
self.assertEqual(4, self.client.delete_all_cached_images())
self.client.do_request.assert_called_with("DELETE", "/cached_images")
def test_queue_image_for_caching(self):
self.client.do_request.return_value = utils.FakeHTTPResponse()
self.assertTrue(self.client.queue_image_for_caching('test_id'))
self.client.do_request.assert_called_with("PUT",
"/queued_images/test_id")
def test_delete_queued_image(self):
self.client.do_request.return_value = utils.FakeHTTPResponse()
self.assertTrue(self.client.delete_queued_image('test_id'))
self.client.do_request.assert_called_with("DELETE",
"/queued_images/test_id")
def test_delete_all_queued_images(self):
expected_data = b'{"num_deleted": 4}'
self.client.do_request.return_value = utils.FakeHTTPResponse(
data=expected_data)
self.assertEqual(4, self.client.delete_all_queued_images())
self.client.do_request.assert_called_with("DELETE", "/queued_images")
class GetClientTestCase(utils.BaseTestCase):
def setUp(self):
super(GetClientTestCase, self).setUp()
self.host = 'test_host'
self.env = os.environ.copy()
os.environ.clear()
def tearDown(self):
os.environ = self.env
super(GetClientTestCase, self).tearDown()
def test_get_client_host_only(self):
expected_creds = {
'username': None,
'password': None,
'tenant': None,
'auth_url': None,
'strategy': 'noauth',
'region': None
}
self.assertEqual(expected_creds, client.get_client(self.host).creds)
def test_get_client_all_creds(self):
expected_creds = {
'username': 'name',
'password': 'pass',
'tenant': 'ten',
'auth_url': 'url',
'strategy': 'keystone',
'region': 'reg'
}
creds = client.get_client(
self.host,
username='name',
password='pass',
tenant='ten',
auth_url='url',
auth_strategy='strategy',
region='reg'
).creds
self.assertEqual(expected_creds, creds)
def test_get_client_using_provided_host(self):
cli = client.get_client(self.host)
cli._do_request = mock.MagicMock()
cli.configure_from_url = mock.MagicMock()
cli.auth_plugin.management_url = mock.MagicMock()
cli.do_request("GET", "/queued_images")
self.assertFalse(cli.configure_from_url.called)
self.assertFalse(client.get_client(self.host).configure_via_auth)
def test_get_client_client_configuration_error(self):
self.assertRaises(exception.ClientConfigurationError,
client.get_client, self.host, username='name',
password='pass', tenant='ten',
auth_strategy='keystone', region='reg')

View File

@ -71,18 +71,6 @@ class VersionsTest(base.IsolatedUnitTest):
'links': [{'rel': 'self',
'href': '%s/v2/' % url}],
},
{
'id': 'v1.1',
'status': 'DEPRECATED',
'links': [{'rel': 'self',
'href': '%s/v1/' % url}],
},
{
'id': 'v1.0',
'status': 'DEPRECATED',
'links': [{'rel': 'self',
'href': '%s/v1/' % url}],
},
]
return versions
@ -142,27 +130,6 @@ class VersionNegotiationTest(base.IsolatedUnitTest):
super(VersionNegotiationTest, self).setUp()
self.middleware = version_negotiation.VersionNegotiationFilter(None)
def test_request_url_v1(self):
request = webob.Request.blank('/v1/images')
self.middleware.process_request(request)
self.assertEqual('/v1/images', request.path_info)
def test_request_url_v1_0(self):
request = webob.Request.blank('/v1.0/images')
self.middleware.process_request(request)
self.assertEqual('/v1/images', request.path_info)
def test_request_url_v1_1(self):
request = webob.Request.blank('/v1.1/images')
self.middleware.process_request(request)
self.assertEqual('/v1/images', request.path_info)
def test_request_accept_v1(self):
request = webob.Request.blank('/images')
request.headers = {'accept': 'application/vnd.openstack.images-v1'}
self.middleware.process_request(request)
self.assertEqual('/v1/images', request.path_info)
def test_request_url_v2(self):
request = webob.Request.blank('/v2/images')
self.middleware.process_request(request)