Merge "Remove use of deprecated optparse module"

This commit is contained in:
Zuul 2017-12-01 08:11:02 +00:00 committed by Gerrit Code Review
commit 9478e1a332
2 changed files with 279 additions and 409 deletions

View File

@ -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:

View File

@ -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)