Merge "Remove use of deprecated optparse module"
This commit is contained in:
commit
9478e1a332
|
@ -20,9 +20,10 @@ A simple cache management utility for Glance.
|
|||
"""
|
||||
from __future__ import print_function
|
||||
|
||||
import argparse
|
||||
import collections
|
||||
import datetime
|
||||
import functools
|
||||
import optparse
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
@ -80,12 +81,12 @@ def catch_error(action):
|
|||
|
||||
|
||||
@catch_error('show cached images')
|
||||
def list_cached(options, args):
|
||||
def list_cached(args):
|
||||
"""%(prog)s list-cached [options]
|
||||
|
||||
List all images currently cached.
|
||||
List all images currently cached.
|
||||
"""
|
||||
client = get_client(options)
|
||||
client = get_client(args)
|
||||
images = client.get_cached_images()
|
||||
if not images:
|
||||
print("No cached images.")
|
||||
|
@ -118,15 +119,16 @@ List all images currently cached.
|
|||
image['hits']))
|
||||
|
||||
print(pretty_table.get_string())
|
||||
return SUCCESS
|
||||
|
||||
|
||||
@catch_error('show queued images')
|
||||
def list_queued(options, args):
|
||||
def list_queued(args):
|
||||
"""%(prog)s list-queued [options]
|
||||
|
||||
List all images currently queued for caching.
|
||||
List all images currently queued for caching.
|
||||
"""
|
||||
client = get_client(options)
|
||||
client = get_client(args)
|
||||
images = client.get_queued_images()
|
||||
if not images:
|
||||
print("No queued images.")
|
||||
|
@ -143,27 +145,27 @@ List all images currently queued for caching.
|
|||
|
||||
|
||||
@catch_error('queue the specified image for caching')
|
||||
def queue_image(options, args):
|
||||
def queue_image(args):
|
||||
"""%(prog)s queue-image <IMAGE_ID> [options]
|
||||
|
||||
Queues an image for caching
|
||||
"""
|
||||
if len(args) == 1:
|
||||
image_id = args.pop()
|
||||
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 options.force and
|
||||
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(options)
|
||||
client = get_client(args)
|
||||
client.queue_image_for_caching(image_id)
|
||||
|
||||
if options.verbose:
|
||||
if args.verbose:
|
||||
print("Queued image %(image_id)s for caching" %
|
||||
{'image_id': image_id})
|
||||
|
||||
|
@ -171,47 +173,46 @@ Queues an image for caching
|
|||
|
||||
|
||||
@catch_error('delete the specified cached image')
|
||||
def delete_cached_image(options, args):
|
||||
"""
|
||||
%(prog)s delete-cached-image <IMAGE_ID> [options]
|
||||
def delete_cached_image(args):
|
||||
"""%(prog)s delete-cached-image <IMAGE_ID> [options]
|
||||
|
||||
Deletes an image from the cache
|
||||
Deletes an image from the cache.
|
||||
"""
|
||||
if len(args) == 1:
|
||||
image_id = args.pop()
|
||||
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 options.force and
|
||||
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(options)
|
||||
client = get_client(args)
|
||||
client.delete_cached_image(image_id)
|
||||
|
||||
if options.verbose:
|
||||
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(options, args):
|
||||
def delete_all_cached_images(args):
|
||||
"""%(prog)s delete-all-cached-images [options]
|
||||
|
||||
Remove all images from the cache.
|
||||
Remove all images from the cache.
|
||||
"""
|
||||
if (not options.force and
|
||||
if (not args.force and
|
||||
not user_confirm("Delete all cached images?", default=False)):
|
||||
return SUCCESS
|
||||
|
||||
client = get_client(options)
|
||||
client = get_client(args)
|
||||
num_deleted = client.delete_all_cached_images()
|
||||
|
||||
if options.verbose:
|
||||
if args.verbose:
|
||||
print("Deleted %(num_deleted)s cached images" %
|
||||
{'num_deleted': num_deleted})
|
||||
|
||||
|
@ -219,47 +220,46 @@ Remove all images from the cache.
|
|||
|
||||
|
||||
@catch_error('delete the specified queued image')
|
||||
def delete_queued_image(options, args):
|
||||
"""
|
||||
%(prog)s delete-queued-image <IMAGE_ID> [options]
|
||||
def delete_queued_image(args):
|
||||
"""%(prog)s delete-queued-image <IMAGE_ID> [options]
|
||||
|
||||
Deletes an image from the cache
|
||||
Deletes an image from the cache.
|
||||
"""
|
||||
if len(args) == 1:
|
||||
image_id = args.pop()
|
||||
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 options.force and
|
||||
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(options)
|
||||
client = get_client(args)
|
||||
client.delete_queued_image(image_id)
|
||||
|
||||
if options.verbose:
|
||||
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(options, args):
|
||||
def delete_all_queued_images(args):
|
||||
"""%(prog)s delete-all-queued-images [options]
|
||||
|
||||
Remove all images from the cache queue.
|
||||
Remove all images from the cache queue.
|
||||
"""
|
||||
if (not options.force and
|
||||
if (not args.force and
|
||||
not user_confirm("Delete all queued images?", default=False)):
|
||||
return SUCCESS
|
||||
|
||||
client = get_client(options)
|
||||
client = get_client(args)
|
||||
num_deleted = client.delete_all_queued_images()
|
||||
|
||||
if options.verbose:
|
||||
if args.verbose:
|
||||
print("Deleted %(num_deleted)s queued images" %
|
||||
{'num_deleted': num_deleted})
|
||||
|
||||
|
@ -298,165 +298,149 @@ def env(*vars, **kwargs):
|
|||
return kwargs.get('default', '')
|
||||
|
||||
|
||||
def create_options(parser):
|
||||
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_option('-v', '--verbose', default=False, action="store_true",
|
||||
help="Print more verbose output.")
|
||||
parser.add_option('-d', '--debug', default=False, action="store_true",
|
||||
help="Print debugging output.")
|
||||
parser.add_option('-H', '--host', metavar="ADDRESS", default="0.0.0.0",
|
||||
help="Address of Glance API host. "
|
||||
"Default: %default.")
|
||||
parser.add_option('-p', '--port', dest="port", metavar="PORT",
|
||||
type=int, default=9292,
|
||||
help="Port the Glance API host listens on. "
|
||||
"Default: %default.")
|
||||
parser.add_option('-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_option('-f', '--force', dest="force", metavar="FORCE",
|
||||
default=False, action="store_true",
|
||||
help="Prevent select actions from requesting "
|
||||
"user confirmation.")
|
||||
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_option('--os-auth-token',
|
||||
dest='os_auth_token',
|
||||
default=env('OS_AUTH_TOKEN'),
|
||||
help='Defaults to env[OS_AUTH_TOKEN].')
|
||||
parser.add_option('-A', '--os_auth_token', '--auth_token',
|
||||
dest='os_auth_token',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('--os-username',
|
||||
dest='os_username',
|
||||
default=env('OS_USERNAME'),
|
||||
help='Defaults to env[OS_USERNAME].')
|
||||
parser.add_option('-I', '--os_username',
|
||||
dest='os_username',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('--os-password',
|
||||
dest='os_password',
|
||||
default=env('OS_PASSWORD'),
|
||||
help='Defaults to env[OS_PASSWORD].')
|
||||
parser.add_option('-K', '--os_password',
|
||||
dest='os_password',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('--os-region-name',
|
||||
dest='os_region_name',
|
||||
default=env('OS_REGION_NAME'),
|
||||
help='Defaults to env[OS_REGION_NAME].')
|
||||
parser.add_option('-R', '--os_region_name',
|
||||
dest='os_region_name',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('--os-tenant-id',
|
||||
dest='os_tenant_id',
|
||||
default=env('OS_TENANT_ID'),
|
||||
help='Defaults to env[OS_TENANT_ID].')
|
||||
parser.add_option('--os_tenant_id',
|
||||
dest='os_tenant_id',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('--os-tenant-name',
|
||||
dest='os_tenant_name',
|
||||
default=env('OS_TENANT_NAME'),
|
||||
help='Defaults to env[OS_TENANT_NAME].')
|
||||
parser.add_option('-T', '--os_tenant_name',
|
||||
dest='os_tenant_name',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('--os-auth-url',
|
||||
default=env('OS_AUTH_URL'),
|
||||
help='Defaults to env[OS_AUTH_URL].')
|
||||
parser.add_option('-N', '--os_auth_url',
|
||||
dest='os_auth_url',
|
||||
help=optparse.SUPPRESS_HELP)
|
||||
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_option('-S', '--os_auth_strategy', dest="os_auth_strategy",
|
||||
metavar="STRATEGY",
|
||||
help="Authentication strategy (keystone or noauth).")
|
||||
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()
|
||||
|
||||
|
||||
def parse_options(parser, cli_args):
|
||||
"""
|
||||
Returns the parsed CLI options, command to run and its arguments, merged
|
||||
with any same-named options found in a configuration file
|
||||
|
||||
:param parser: The option parser
|
||||
"""
|
||||
if not cli_args:
|
||||
cli_args.append('-h') # Show options in usage output...
|
||||
|
||||
(options, args) = parser.parse_args(cli_args)
|
||||
|
||||
# HACK(sirp): Make the parser available to the print_help method
|
||||
# print_help is a command, so it only accepts (options, args); we could
|
||||
# one-off have it take (parser, options, args), however, for now, I think
|
||||
# this little hack will suffice
|
||||
options.__parser = parser
|
||||
|
||||
if not args:
|
||||
parser.print_usage()
|
||||
sys.exit(0)
|
||||
|
||||
command_name = args.pop(0)
|
||||
command = lookup_command(parser, command_name)
|
||||
|
||||
return (options, command, 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 print_help(options, args):
|
||||
"""
|
||||
Print help specific to a command
|
||||
"""
|
||||
parser = options.__parser
|
||||
def _format_command_help():
|
||||
"""Formats the help string for subcommands."""
|
||||
help_msg = "Commands:\n\n"
|
||||
|
||||
if not args:
|
||||
parser.print_help()
|
||||
else:
|
||||
number_of_commands = len(args)
|
||||
if number_of_commands == 1:
|
||||
command_name = args.pop()
|
||||
command = lookup_command(parser, command_name)
|
||||
print(command.__doc__ % {'prog': os.path.basename(sys.argv[0])})
|
||||
else:
|
||||
sys.exit("Please specify one command")
|
||||
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(parser, command_name):
|
||||
BASE_COMMANDS = {'help': print_help}
|
||||
|
||||
CACHE_COMMANDS = {
|
||||
'list-cached': list_cached,
|
||||
'list-queued': list_queued,
|
||||
'queue-image': queue_image,
|
||||
'delete-cached-image': delete_cached_image,
|
||||
'delete-all-cached-images': delete_all_cached_images,
|
||||
'delete-queued-image': delete_queued_image,
|
||||
'delete-all-queued-images': delete_all_queued_images,
|
||||
}
|
||||
|
||||
commands = {}
|
||||
for command_set in (BASE_COMMANDS, CACHE_COMMANDS):
|
||||
commands.update(command_set)
|
||||
|
||||
def lookup_command(command_name):
|
||||
try:
|
||||
command = commands[command_name]
|
||||
command = CACHE_COMMANDS[command_name]
|
||||
return command[0]
|
||||
except KeyError:
|
||||
parser.print_usage()
|
||||
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})
|
||||
|
||||
return command
|
||||
|
||||
|
||||
def user_confirm(prompt, default=False):
|
||||
"""Yes/No question dialog with user.
|
||||
|
@ -480,39 +464,23 @@ def user_confirm(prompt, default=False):
|
|||
|
||||
|
||||
def main():
|
||||
usage = """
|
||||
%prog <command> [options] [args]
|
||||
parser = argparse.ArgumentParser(
|
||||
description=_format_command_help(),
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
args = parse_args(parser)
|
||||
|
||||
Commands:
|
||||
if args.command[0] == 'help' and len(args.command) == 1:
|
||||
parser.print_help()
|
||||
return
|
||||
|
||||
help <command> Output help for one of the commands below
|
||||
|
||||
list-cached List all images currently cached
|
||||
|
||||
list-queued List all images currently queued for caching
|
||||
|
||||
queue-image Queue an image for caching
|
||||
|
||||
delete-cached-image Purges an image from the cache
|
||||
|
||||
delete-all-cached-images Removes all images from the cache
|
||||
|
||||
delete-queued-image Deletes an image from the cache queue
|
||||
|
||||
delete-all-queued-images Deletes all images from the cache queue
|
||||
"""
|
||||
|
||||
version_string = version.cached_version_string()
|
||||
oparser = optparse.OptionParser(version=version_string,
|
||||
usage=usage.strip())
|
||||
create_options(oparser)
|
||||
(options, command, args) = parse_options(oparser, sys.argv[1:])
|
||||
# Look up the command to run
|
||||
command = lookup_command(args.command[0])
|
||||
|
||||
try:
|
||||
start_time = time.time()
|
||||
result = command(options, args)
|
||||
result = command(args)
|
||||
end_time = time.time()
|
||||
if options.verbose:
|
||||
if args.verbose:
|
||||
print("Completed in %-0.4f sec." % (end_time - start_time))
|
||||
sys.exit(result)
|
||||
except (RuntimeError, NotImplementedError) as e:
|
||||
|
|
|
@ -10,12 +10,11 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import optparse
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
import mock
|
||||
import prettytable
|
||||
from six.moves import StringIO
|
||||
|
||||
from glance.cmd import cache_manage
|
||||
from glance.common import exception
|
||||
|
@ -27,39 +26,40 @@ from glance.tests import utils as test_utils
|
|||
@mock.patch('sys.stdout', mock.Mock())
|
||||
class TestGlanceCmdManage(test_utils.BaseTestCase):
|
||||
|
||||
@mock.patch.object(optparse.OptionParser, 'print_help')
|
||||
@mock.patch.object(optparse.OptionParser, 'parse_args')
|
||||
def test_help(self, mock_parse_args, mock_print_help):
|
||||
mock_parse_args.return_value = (optparse.Values(), ['help'])
|
||||
oparser = optparse.OptionParser()
|
||||
(options, command, args) = cache_manage.parse_options(oparser,
|
||||
['help'])
|
||||
command(options, args)
|
||||
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(optparse.OptionParser, 'parse_args')
|
||||
def test_help_with_command(self, mock_parse_args):
|
||||
mock_parse_args.return_value = (optparse.Values(), ['help',
|
||||
'list-cached'])
|
||||
oparser = optparse.OptionParser()
|
||||
(options, command, args) = cache_manage.parse_options(oparser,
|
||||
['help',
|
||||
'list-cached'])
|
||||
command(options, args)
|
||||
@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')
|
||||
|
||||
@mock.patch.object(sys, 'exit')
|
||||
@mock.patch.object(optparse.OptionParser, 'parse_args')
|
||||
def test_help_with_redundant_command(self, mock_parse_args, mock_exit):
|
||||
mock_parse_args.return_value = (optparse.Values(), ['help',
|
||||
'list-cached',
|
||||
"1"])
|
||||
oparser = optparse.OptionParser()
|
||||
(options, command, args) = cache_manage.parse_options(oparser,
|
||||
['help',
|
||||
'list-cached',
|
||||
"1"])
|
||||
command(options, args)
|
||||
self.assertEqual(1, mock_exit.call_count)
|
||||
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')
|
||||
|
@ -69,7 +69,6 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
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),
|
||||
|
@ -77,8 +76,7 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
{'last_accessed': float(1378985797.124511),
|
||||
'last_modified': float(1378985797.124511),
|
||||
'image_id': '2', 'size': '255', 'hits': '2'}]
|
||||
cache_manage.list_cached(mock.Mock(), '')
|
||||
|
||||
self._run_command(['list-cached'], cache_manage.SUCCESS)
|
||||
self.assertEqual(len(mock_images.return_value),
|
||||
mock_row_create.call_count)
|
||||
|
||||
|
@ -89,10 +87,7 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
Verify that list_cached() method handles a case when no images are
|
||||
cached without errors.
|
||||
"""
|
||||
|
||||
mock_images.return_value = []
|
||||
self.assertEqual(cache_manage.SUCCESS,
|
||||
cache_manage.list_cached(mock.Mock(), ''))
|
||||
self._run_command(['list-cached'], cache_manage.SUCCESS)
|
||||
|
||||
@mock.patch.object(glance.image_cache.client.CacheClient,
|
||||
'get_queued_images')
|
||||
|
@ -102,8 +97,8 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
|
||||
mock_images.return_value = [
|
||||
{'image_id': '1'}, {'image_id': '2'}]
|
||||
cache_manage.list_queued(mock.Mock(), '')
|
||||
|
||||
# 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)
|
||||
|
||||
|
@ -114,212 +109,161 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
Verify that list_queued() method handles a case when no images were
|
||||
queued without errors.
|
||||
"""
|
||||
|
||||
mock_images.return_value = []
|
||||
|
||||
self.assertEqual(cache_manage.SUCCESS,
|
||||
cache_manage.list_queued(mock.Mock(), ''))
|
||||
self._run_command(['list-queued'], cache_manage.SUCCESS)
|
||||
|
||||
def test_queue_image_without_index(self):
|
||||
self.assertEqual(cache_manage.FAILURE,
|
||||
cache_manage.queue_image(mock.Mock(), []))
|
||||
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):
|
||||
# options.forced set to False and queue confirmation set to False.
|
||||
|
||||
# --force not set and queue confirmation return False.
|
||||
mock_confirm.return_value = False
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
self.assertEqual(cache_manage.SUCCESS,
|
||||
cache_manage.queue_image(mock_options, ['img_id']))
|
||||
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_client, mock_confirm):
|
||||
# options.forced set to False and queue confirmation set to True.
|
||||
|
||||
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_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
mock_options.verbose = True # to cover additional condition and line
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(mock_client, 'mock_client')
|
||||
mock_client = mock.MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
self.assertEqual(cache_manage.SUCCESS,
|
||||
cache_manage.queue_image(mock_options, ['img_id']))
|
||||
self.assertTrue(mock_client.called)
|
||||
self.assertIn(
|
||||
mock.call.mock_client().queue_image_for_caching('img_id'),
|
||||
manager.mock_calls)
|
||||
# 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.assertEqual(cache_manage.FAILURE,
|
||||
cache_manage.delete_cached_image(mock.Mock(), []))
|
||||
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):
|
||||
# options.forced set to False and delete confirmation set to False.
|
||||
|
||||
# --force not set and confirmation return False.
|
||||
mock_confirm.return_value = False
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_cached_image(mock_options, ['img_id']))
|
||||
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_client,
|
||||
def test_delete_cached_image_not_forced_confirmed(self, mock_get_client,
|
||||
mock_confirm):
|
||||
# options.forced set to False and delete confirmation set to True.
|
||||
|
||||
# --force not set and confirmation return True.
|
||||
mock_confirm.return_value = True
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
mock_options.verbose = True # to cover additional condition and line
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(mock_client, 'mock_client')
|
||||
mock_client = mock.MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_cached_image(mock_options, ['img_id']))
|
||||
# verbose to cover additional condition and line
|
||||
self._run_command(['delete-cached-image', 'fakeimageid', '-v'],
|
||||
cache_manage.SUCCESS)
|
||||
|
||||
self.assertIn(
|
||||
mock.call.mock_client().delete_cached_image('img_id'),
|
||||
manager.mock_calls)
|
||||
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):
|
||||
# options.forced set to False and delete confirmation set to False.
|
||||
|
||||
# --force not set and confirmation return False.
|
||||
mock_confirm.return_value = False
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_all_cached_images(mock_options, None))
|
||||
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_client,
|
||||
def test_delete_cached_images_not_forced_confirmed(self, mock_get_client,
|
||||
mock_confirm):
|
||||
# options.forced set to False and delete confirmation set to True.
|
||||
|
||||
# --force not set and confirmation return True.
|
||||
mock_confirm.return_value = True
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
mock_options.verbose = True # to cover additional condition and line
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(mock_client, 'mock_client')
|
||||
mock_client = mock.MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_all_cached_images(mock_options, None))
|
||||
self.assertTrue(mock_client.called)
|
||||
self.assertIn(
|
||||
mock.call.mock_client().delete_all_cached_images(),
|
||||
manager.mock_calls)
|
||||
# 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.assertEqual(cache_manage.FAILURE,
|
||||
cache_manage.delete_queued_image(mock.Mock(), []))
|
||||
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):
|
||||
# options.forced set to False and delete confirmation set to False.
|
||||
|
||||
# --force not set and confirmation set to False.
|
||||
mock_confirm.return_value = False
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_queued_image(mock_options, ['img_id']))
|
||||
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_client,
|
||||
def test_delete_queued_image_not_forced_confirmed(self, mock_get_client,
|
||||
mock_confirm):
|
||||
# options.forced set to False and delete confirmation set to True.
|
||||
|
||||
# --force not set and confirmation set to True.
|
||||
mock_confirm.return_value = True
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
mock_options.verbose = True # to cover additional condition and line
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(mock_client, 'mock_client')
|
||||
mock_client = mock.MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_queued_image(mock_options, ['img_id']))
|
||||
self.assertTrue(mock_client.called)
|
||||
self.assertIn(
|
||||
mock.call.mock_client().delete_queued_image('img_id'),
|
||||
manager.mock_calls)
|
||||
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):
|
||||
# options.forced set to False and delete confirmation set to False.
|
||||
|
||||
# --force not set and confirmation set to False.
|
||||
mock_confirm.return_value = False
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_all_queued_images(mock_options, None))
|
||||
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_client,
|
||||
def test_delete_queued_images_not_forced_confirmed(self, mock_get_client,
|
||||
mock_confirm):
|
||||
# options.forced set to False and delete confirmation set to True.
|
||||
# --force not set and confirmation set to True.
|
||||
mock_confirm.return_value = True
|
||||
mock_options = mock.Mock()
|
||||
mock_options.force = False
|
||||
mock_options.verbose = True # to cover additional condition and line
|
||||
manager = mock.MagicMock()
|
||||
manager.attach_mock(mock_client, 'mock_client')
|
||||
mock_client = mock.MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
self.assertEqual(
|
||||
cache_manage.SUCCESS,
|
||||
cache_manage.delete_all_queued_images(mock_options, None))
|
||||
self.assertTrue(mock_client.called)
|
||||
self.assertIn(
|
||||
mock.call.mock_client().delete_all_queued_images(),
|
||||
manager.mock_calls)
|
||||
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(), None))
|
||||
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(), None))
|
||||
cache_manage.list_cached(mock.Mock()))
|
||||
|
||||
@mock.patch.object(glance.cmd.cache_manage, 'get_client')
|
||||
def test_catch_error_unhandled(self, mock_function):
|
||||
|
@ -328,7 +272,7 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
my_mock.debug = False
|
||||
|
||||
self.assertEqual(cache_manage.FAILURE,
|
||||
cache_manage.list_cached(my_mock, None))
|
||||
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):
|
||||
|
@ -337,7 +281,7 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
my_mock.debug = True
|
||||
|
||||
self.assertRaises(exception.Duplicate,
|
||||
cache_manage.list_cached, my_mock, None)
|
||||
cache_manage.list_cached, my_mock)
|
||||
|
||||
def test_cache_manage_env(self):
|
||||
def_value = 'sometext12345678900987654321'
|
||||
|
@ -350,47 +294,5 @@ class TestGlanceCmdManage(test_utils.BaseTestCase):
|
|||
cache_manage.env('TMPVALUE1234567890',
|
||||
default=def_value))
|
||||
|
||||
def test_create_option(self):
|
||||
oparser = optparse.OptionParser()
|
||||
cache_manage.create_options(oparser)
|
||||
self.assertGreater(len(oparser.option_list), 0)
|
||||
|
||||
@mock.patch.object(glance.cmd.cache_manage, 'lookup_command')
|
||||
def test_parse_options_no_parameters(self, mock_lookup):
|
||||
with mock.patch('sys.stdout', new_callable=StringIO):
|
||||
oparser = optparse.OptionParser()
|
||||
cache_manage.create_options(oparser)
|
||||
|
||||
result = self.assertRaises(SystemExit, cache_manage.parse_options,
|
||||
oparser, [])
|
||||
self.assertEqual(0, result.code)
|
||||
self.assertFalse(mock_lookup.called)
|
||||
|
||||
@mock.patch.object(optparse.OptionParser, 'print_usage')
|
||||
def test_parse_options_no_arguments(self, mock_printout):
|
||||
oparser = optparse.OptionParser()
|
||||
cache_manage.create_options(oparser)
|
||||
|
||||
result = self.assertRaises(SystemExit, cache_manage.parse_options,
|
||||
oparser, ['-p', '1212'])
|
||||
self.assertEqual(0, result.code)
|
||||
self.assertTrue(mock_printout.called)
|
||||
|
||||
@mock.patch.object(glance.cmd.cache_manage, 'lookup_command')
|
||||
def test_parse_options_retrieve_command(self, mock_lookup):
|
||||
mock_lookup.return_value = True
|
||||
oparser = optparse.OptionParser()
|
||||
cache_manage.create_options(oparser)
|
||||
(options, command, args) = cache_manage.parse_options(oparser,
|
||||
['-p', '1212',
|
||||
'list-cached'])
|
||||
|
||||
self.assertTrue(command)
|
||||
|
||||
def test_lookup_command_unsupported_command(self):
|
||||
self.assertRaises(SystemExit, cache_manage.lookup_command, mock.Mock(),
|
||||
'unsupported_command')
|
||||
|
||||
def test_lookup_command_supported_command(self):
|
||||
command = cache_manage.lookup_command(mock.Mock(), 'list-cached')
|
||||
self.assertEqual(cache_manage.list_cached, command)
|
||||
self._run_command(['unsupported_command'], cache_manage.FAILURE)
|
||||
|
|
Loading…
Reference in New Issue