openstack/python-openstackclient
Code Issues Proposed changes

Blacken openstack.common

Browse Source 
Black used with the '-l 79 -S' flags.

A future change will ignore this commit in git-blame history by adding a
'git-blame-ignore-revs' file.

Change-Id: Ifcb3c798666d74d596b8ecb3d6d507f782de7ba5
Signed-off-by: Stephen Finucane <sfinucan@redhat.com>
This commit is contained in:
Stephen Finucane 2023-05-08 11:35:45 +01:00
parent 7ca43885c2
commit 7d80f9e962
29 changed files with 777 additions and 566 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
openstackclient/common/availability_zone.py
View File

@ -28,8 +28,9 @@ LOG = logging.getLogger(__name__)
def _xform_common_availability_zone(az, zone_info):
if hasattr(az, 'zoneState'):
zone_info['zone_status'] = ('available' if az.zoneState['available']
else 'not available')
zone_info['zone_status'] = (
'available' if az.zoneState['available'] else 'not available'
)
if hasattr(az, 'zoneName'):
zone_info['zone_name'] = az.zoneName
@ -56,7 +57,8 @@ def _xform_compute_availability_zone(az, include_extra):
info['service_status'] = '%s %s %s' % (
'enabled' if state['active'] else 'disabled',
':-)' if state['available'] else 'XXX',
state['updated_at'])
state['updated_at'],
)
result.append(info)
else:
zone_info['host_name'] = ''
@ -141,8 +143,10 @@ class ListAvailabilityZone(command.Lister):
except Exception as e:
LOG.debug('Volume availability zone exception: %s', e)
if parsed_args.volume:
message = _("Availability zones list not supported by "
"Block Storage API")
message = _(
"Availability zones list not supported by "
"Block Storage API"
)
LOG.warning(message)
result = []
@ -154,13 +158,15 @@ class ListAvailabilityZone(command.Lister):
network_client = self.app.client_manager.network
try:
# Verify that the extension exists.
network_client.find_extension('Availability Zone',
ignore_missing=False)
network_client.find_extension(
'Availability Zone', ignore_missing=False
)
except Exception as e:
LOG.debug('Network availability zone exception: ', e)
if parsed_args.network:
message = _("Availability zones list not supported by "
"Network API")
message = _(
"Availability zones list not supported by " "Network API"
)
LOG.warning(message)
return []
@ -170,17 +176,24 @@ class ListAvailabilityZone(command.Lister):
return result
def take_action(self, parsed_args):
if parsed_args.long:
columns = ('Zone Name', 'Zone Status', 'Zone Resource',
'Host Name', 'Service Name', 'Service Status')
columns = (
'Zone Name',
'Zone Status',
'Zone Resource',
'Host Name',
'Service Name',
'Service Status',
)
else:
columns = ('Zone Name', 'Zone Status')
# Show everything by default.
show_all = (not parsed_args.compute and
not parsed_args.volume and
not parsed_args.network)
show_all = (
not parsed_args.compute
and not parsed_args.volume
and not parsed_args.network
)
result = []
if parsed_args.compute or show_all:
@ -190,7 +203,7 @@ class ListAvailabilityZone(command.Lister):
if parsed_args.network or show_all:
result += self._get_network_availability_zones(parsed_args)
return (columns,
(utils.get_dict_properties(
s, columns
) for s in result))
return (
columns,
(utils.get_dict_properties(s, columns) for s in result),
)

34
openstackclient/common/clientmanager.py
View File

@ -77,18 +77,20 @@ class ClientManager(clientmanager.ClientManager):
# CloudConfig.__init__() and we'll die if it was not
# passed.
if (
self._auth_required and
self._cli_options._openstack_config is not None
self._auth_required
and self._cli_options._openstack_config is not None
):
self._cli_options._openstack_config._pw_callback = \
self._cli_options._openstack_config._pw_callback = (
shell.prompt_for_password
)
try:
# We might already get auth from SDK caching
if not self._cli_options._auth:
self._cli_options._auth = \
self._cli_options._auth = (
self._cli_options._openstack_config.load_auth_plugin(
self._cli_options.config,
)
)
except TypeError as e:
self._fallback_load_auth_plugin(e)
@ -101,14 +103,14 @@ class ClientManager(clientmanager.ClientManager):
# We know it looks ugly, but it's necessary.
if self._cli_options.config['auth']['token'] == 'x':
# restore original auth_type
self._cli_options.config['auth_type'] = \
self._original_auth_type
self._cli_options.config['auth_type'] = self._original_auth_type
del self._cli_options.config['auth']['token']
del self._cli_options.config['auth']['endpoint']
self._cli_options._auth = \
self._cli_options._auth = (
self._cli_options._openstack_config.load_auth_plugin(
self._cli_options.config,
)
)
else:
raise e
@ -132,8 +134,10 @@ class ClientManager(clientmanager.ClientManager):
# name so we need to figure out which version to look
# for when calling is_service_available()
volume_version = volume_client.api_version.ver_major
if self.is_service_available(
"volumev%s" % volume_version) is not False:
if (
self.is_service_available("volumev%s" % volume_version)
is not False
):
return True
elif self.is_service_available('volume') is not False:
return True
@ -143,6 +147,7 @@ class ClientManager(clientmanager.ClientManager):
# Plugin Support
def get_plugin_modules(group):
"""Find plugin entry points"""
mod_list = []
@ -165,7 +170,8 @@ def get_plugin_modules(group):
module = importlib.import_module(module_name)
except Exception as err:
sys.stderr.write(
"WARNING: Failed to import plugin %s: %s.\n" % (ep.name, err))
"WARNING: Failed to import plugin %s: %s.\n" % (ep.name, err)
)
continue
mod_list.append(module)
@ -198,6 +204,8 @@ PLUGIN_MODULES = get_plugin_modules(
'openstack.cli.base',
)
# Append list of external plugin modules
PLUGIN_MODULES.extend(get_plugin_modules(
'openstack.cli.extension',
))
PLUGIN_MODULES.extend(
get_plugin_modules(
'openstack.cli.extension',
)
)

3
openstackclient/common/configuration.py
View File

@ -53,7 +53,8 @@ class ShowConfiguration(command.ShowOne):
if getattr(self.app.client_manager, "auth_plugin_name", None):
auth_plg_name = self.app.client_manager.auth_plugin_name
secret_opts = [
o.dest for o in base.get_plugin_options(auth_plg_name)
o.dest
for o in base.get_plugin_options(auth_plg_name)
if o.secret
]

41
openstackclient/common/extension.py
View File

@ -65,8 +65,14 @@ class ListExtension(command.Lister):
def take_action(self, parsed_args):
if parsed_args.long:
columns = ('Name', 'Alias', 'Description',
'Namespace', 'Updated', 'Links')
columns = (
'Name',
'Alias',
'Description',
'Namespace',
'Updated',
'Links',
)
else:
columns = ('Name', 'Alias', 'Description')
@ -75,10 +81,12 @@ class ListExtension(command.Lister):
# by default we want to show everything, unless the
# user specifies one or more of the APIs to show
# for now, only identity and compute are supported.
show_all = (not parsed_args.identity and
not parsed_args.compute and
not parsed_args.volume and
not parsed_args.network)
show_all = (
not parsed_args.identity
and not parsed_args.compute
and not parsed_args.volume
and not parsed_args.network
)
if parsed_args.identity or show_all:
identity_client = self.app.client_manager.identity
@ -101,8 +109,9 @@ class ListExtension(command.Lister):
try:
data += volume_client.list_extensions.show_all()
except Exception:
message = _("Extensions list not supported by "
"Block Storage API")
message = _(
"Extensions list not supported by " "Block Storage API"
)
LOG.warning(message)
if parsed_args.network or show_all:
@ -110,15 +119,17 @@ class ListExtension(command.Lister):
try:
data += network_client.extensions()
except Exception:
message = _("Failed to retrieve extensions list "
"from Network API")
message = _(
"Failed to retrieve extensions list " "from Network API"
)
LOG.warning(message)
extension_tuples = (
utils.get_item_properties(
s,
columns,
) for s in data
)
for s in data
)
return (columns, extension_tuples)
@ -132,9 +143,11 @@ class ShowExtension(command.ShowOne):
parser.add_argument(
'extension',
metavar='<extension>',
help=_('Extension to display. '
'Currently, only network extensions are supported. '
'(Name or Alias)'),
help=_(
'Extension to display. '
'Currently, only network extensions are supported. '
'(Name or Alias)'
),
)
return parser

64
openstackclient/common/limits.py
View File

@ -54,19 +54,22 @@ class ShowLimits(command.Lister):
parser.add_argument(
'--project',
metavar='<project>',
help=_('Show limits for a specific project (name or ID)'
' [only valid with --absolute]'),
help=_(
'Show limits for a specific project (name or ID)'
' [only valid with --absolute]'
),
)
parser.add_argument(
'--domain',
metavar='<domain>',
help=_('Domain the project belongs to (name or ID)'
' [only valid with --absolute]'),
help=_(
'Domain the project belongs to (name or ID)'
' [only valid with --absolute]'
),
)
return parser
def take_action(self, parsed_args):
compute_client = self.app.client_manager.compute
volume_client = self.app.client_manager.volume
@ -74,21 +77,26 @@ class ShowLimits(command.Lister):
if parsed_args.project is not None:
identity_client = self.app.client_manager.identity
if parsed_args.domain is not None:
domain = identity_common.find_domain(identity_client,
parsed_args.domain)
project_id = utils.find_resource(identity_client.projects,
parsed_args.project,
domain_id=domain.id).id
domain = identity_common.find_domain(
identity_client, parsed_args.domain
)
project_id = utils.find_resource(
identity_client.projects,
parsed_args.project,
domain_id=domain.id,
).id
else:
project_id = utils.find_resource(identity_client.projects,
parsed_args.project).id
project_id = utils.find_resource(
identity_client.projects, parsed_args.project
).id
compute_limits = None
volume_limits = None
if self.app.client_manager.is_compute_endpoint_enabled():
compute_limits = compute_client.limits.get(parsed_args.is_reserved,
tenant_id=project_id)
compute_limits = compute_client.limits.get(
parsed_args.is_reserved, tenant_id=project_id
)
if self.app.client_manager.is_volume_endpoint_enabled(volume_client):
volume_limits = volume_client.limits.get()
@ -100,17 +108,33 @@ class ShowLimits(command.Lister):
if volume_limits:
data.append(volume_limits.absolute)
columns = ["Name", "Value"]
return (columns, (utils.get_item_properties(s, columns)
for s in itertools.chain(*data)))
return (
columns,
(
utils.get_item_properties(s, columns)
for s in itertools.chain(*data)
),
)
elif parsed_args.is_rate:
if compute_limits:
data.append(compute_limits.rate)
if volume_limits:
data.append(volume_limits.rate)
columns = ["Verb", "URI", "Value", "Remain", "Unit",
"Next Available"]
return (columns, (utils.get_item_properties(s, columns)
for s in itertools.chain(*data)))
columns = [
"Verb",
"URI",
"Value",
"Remain",
"Unit",
"Next Available",
]
return (
columns,
(
utils.get_item_properties(s, columns)
for s in itertools.chain(*data)
),
)
else:
return {}, {}

19
openstackclient/common/module.py
View File

@ -33,9 +33,11 @@ class ListCommand(command.Lister):
parser.add_argument(
'--group',
metavar='<group-keyword>',
help=_('Show commands filtered by a command group, for example: '
'identity, volume, compute, image, network and '
'other keywords'),
help=_(
'Show commands filtered by a command group, for example: '
'identity, volume, compute, image, network and '
'other keywords'
),
)
return parser
@ -54,7 +56,6 @@ class ListCommand(command.Lister):
command_names = sorted(command_names)
if command_names != []:
# TODO(bapalm): Fix this when cliff properly supports
# handling the detection rather than using the hard-code below.
if parsed_args.formatter == 'table':
@ -81,7 +82,6 @@ class ListModule(command.ShowOne):
return parser
def take_action(self, parsed_args):
data = {}
# Get module versions
mods = sys.modules
@ -95,9 +95,12 @@ class ListModule(command.ShowOne):
# show for the default (not --all) invocation.
# It should be just the things we actually care
# about like client and plugin modules...
if (parsed_args.all or
# Handle xxxclient and openstacksdk
(k.endswith('client') or k == 'openstack')):
if (
parsed_args.all
or
# Handle xxxclient and openstacksdk
(k.endswith('client') or k == 'openstack')
):
try:
# NOTE(RuiChen): openstacksdk bug/1588823 exist,
# no good way to add __version__ for

8
openstackclient/common/progressbar.py
View File

@ -38,9 +38,11 @@ class _ProgressBarBase(object):
if self._show_progress:
self._percent += size_read / self._totalsize
# Output something like this: [==========> ] 49%
sys.stdout.write('\r[{0:<30}] {1:.0%}'.format(
'=' * int(round(self._percent * 29)) + '>', self._percent
))
sys.stdout.write(
'\r[{0:<30}] {1:.0%}'.format(
'=' * int(round(self._percent * 29)) + '>', self._percent
)
)
sys.stdout.flush()
def __getattr__(self, attr):

52
openstackclient/common/project_cleanup.py
View File

@ -35,8 +35,7 @@ def ask_user_yesno(msg):
:return bool: User choice
"""
while True:
answer = getpass._raw_input(
'{} [{}]: '.format(msg, 'y/n'))
answer = getpass._raw_input('{} [{}]: '.format(msg, 'y/n'))
if answer in ('y', 'Y', 'yes'):
return True
elif answer in ('n', 'N', 'no'):
@ -52,33 +51,33 @@ class ProjectCleanup(command.Command):
action_group.add_argument(
'--dry-run',
action='store_true',
help=_("List a project's resources but do not delete them")
help=_("List a project's resources but do not delete them"),
)
action_group.add_argument(
'--auto-approve',
action='store_true',
help=_("Delete resources without asking for confirmation")
help=_("Delete resources without asking for confirmation"),
)
project_group = parser.add_mutually_exclusive_group(required=True)
project_group.add_argument(
'--auth-project',
action='store_true',
help=_('Delete resources of the project used to authenticate')
help=_('Delete resources of the project used to authenticate'),
)
project_group.add_argument(
'--project',
metavar='<project>',
help=_('Project to clean (name or ID)')
help=_('Project to clean (name or ID)'),
)
parser.add_argument(
'--created-before',
metavar='<YYYY-MM-DDTHH24:MI:SS>',
help=_('Only delete resources created before the given time')
help=_('Only delete resources created before the given time'),
)
parser.add_argument(
'--updated-before',
metavar='<YYYY-MM-DDTHH24:MI:SS>',
help=_('Only delete resources updated before the given time')
help=_('Only delete resources updated before the given time'),
)
identity_common.add_project_domain_option_to_parser(parser)
return parser
@ -90,16 +89,18 @@ class ProjectCleanup(command.Command):
project_connect = sdk
elif parsed_args.project:
project = sdk.identity.find_project(
name_or_id=parsed_args.project,
ignore_missing=False)
name_or_id=parsed_args.project, ignore_missing=False
)
project_connect = sdk.connect_as_project(project)
if project_connect:
status_queue = queue.Queue()
parsed_args.max_width = int(os.environ.get('CLIFF_MAX_TERM_WIDTH',
0))
parsed_args.fit_width = bool(int(os.environ.get('CLIFF_FIT_WIDTH',
0)))
parsed_args.max_width = int(
os.environ.get('CLIFF_MAX_TERM_WIDTH', 0)
)
parsed_args.fit_width = bool(
int(os.environ.get('CLIFF_FIT_WIDTH', 0))
)
parsed_args.print_empty = False
table_fmt = table.TableFormatter()
@ -112,22 +113,20 @@ class ProjectCleanup(command.Command):
if parsed_args.updated_before:
filters['updated_at'] = parsed_args.updated_before
project_connect.project_cleanup(dry_run=True,
status_queue=status_queue,
filters=filters)
project_connect.project_cleanup(
dry_run=True, status_queue=status_queue, filters=filters
)
data = []
while not status_queue.empty():
resource = status_queue.get_nowait()
data.append(
(type(resource).__name__, resource.id, resource.name))
(type(resource).__name__, resource.id, resource.name)
)
status_queue.task_done()
status_queue.join()
table_fmt.emit_list(
('Type', 'ID', 'Name'),
data,
self.app.stdout,
parsed_args
('Type', 'ID', 'Name'), data, self.app.stdout, parsed_args
)
if parsed_args.dry_run:
@ -135,11 +134,12 @@ class ProjectCleanup(command.Command):
if not parsed_args.auto_approve:
if not ask_user_yesno(
_("These resources will be deleted. Are you sure")):
_("These resources will be deleted. Are you sure")
):
return
self.log.warning(_('Deleting resources'))
project_connect.project_cleanup(dry_run=False,
status_queue=status_queue,
filters=filters)
project_connect.project_cleanup(
dry_run=False, status_queue=status_queue, filters=filters
)

42
openstackclient/common/project_purge.py
View File

@ -88,7 +88,8 @@ class ProjectPurge(command.Command):
search_opts = {'tenant_id': project_id, 'all_tenants': True}
data = compute_client.servers.list(search_opts=search_opts)
self.delete_objects(
compute_client.servers.delete, data, 'server', dry_run)
compute_client.servers.delete, data, 'server', dry_run
)
except Exception:
pass
@ -104,7 +105,8 @@ class ProjectPurge(command.Command):
else:
raise NotImplementedError
self.delete_objects(
image_client.images.delete, data, 'image', dry_run)
image_client.images.delete, data, 'image', dry_run
)
except Exception:
pass
@ -117,45 +119,49 @@ class ProjectPurge(command.Command):
self.delete_one_volume_snapshot,
data,
'volume snapshot',
dry_run)
dry_run,
)
except Exception:
pass
try:
data = volume_client.backups.list(search_opts=search_opts)
self.delete_objects(
self.delete_one_volume_backup,
data,
'volume backup',
dry_run)
self.delete_one_volume_backup, data, 'volume backup', dry_run
)
except Exception:
pass
try:
data = volume_client.volumes.list(search_opts=search_opts)
self.delete_objects(
volume_client.volumes.force_delete, data, 'volume', dry_run)
volume_client.volumes.force_delete, data, 'volume', dry_run
)
except Exception:
pass
def delete_objects(self, func_delete, data, resource, dry_run):
result = 0
for i in data:
LOG.warning(_('Deleting %(resource)s : %(id)s') %
{'resource': resource, 'id': i.id})
LOG.warning(
_('Deleting %(resource)s : %(id)s')
% {'resource': resource, 'id': i.id}
)
if not dry_run:
try:
func_delete(i.id)
except Exception as e:
result += 1
LOG.error(_("Failed to delete %(resource)s with "
"ID '%(id)s': %(e)s")
% {'resource': resource, 'id': i.id, 'e': e})
LOG.error(
_(
"Failed to delete %(resource)s with "
"ID '%(id)s': %(e)s"
)
% {'resource': resource, 'id': i.id, 'e': e}
)
if result > 0:
total = len(data)
msg = (_("%(result)s of %(total)s %(resource)ss failed "
"to delete.") %
{'result': result,
'total': total,
'resource': resource})
msg = _(
"%(result)s of %(total)s %(resource)ss failed " "to delete."
) % {'result': result, 'total': total, 'resource': resource}
LOG.error(msg)
def delete_one_volume_snapshot(self, snapshot_id):

6
openstackclient/common/quota.py
View File

@ -749,10 +749,12 @@ class SetQuota(common.NetDetectionMixin, command.Command):
class ShowQuota(command.Lister):
_description = _("""Show quotas for project or class.
_description = _(
"""Show quotas for project or class.
Specify ``--os-compute-api-version 2.50`` or higher to see ``server-groups``
and ``server-group-members`` output for a given quota class.""")
and ``server-group-members`` output for a given quota class."""
)
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)

42
openstackclient/common/versions.py
View File

@ -46,26 +46,29 @@ class ShowVersions(command.Lister):
parser.add_argument(
'--service',
metavar='<service>',
help=_('Show versions for a specific service. The argument should '
'be either an exact match to what is in the catalog or a '
'known official value or alias from '
'service-types-authority '
'(https://service-types.openstack.org/)'),
help=_(
'Show versions for a specific service. The argument should '
'be either an exact match to what is in the catalog or a '
'known official value or alias from '
'service-types-authority '
'(https://service-types.openstack.org/)'
),
)
parser.add_argument(
'--status',
metavar='<status>',
help=_("""Show versions for a specific status. Valid values are:
help=_(
"""Show versions for a specific status. Valid values are:
- SUPPORTED
- CURRENT
- DEPRECATED
- EXPERIMENTAL""")
- EXPERIMENTAL"""
),
)
return parser
def take_action(self, parsed_args):
interface = parsed_args.interface
if parsed_args.is_all_interfaces:
interface = None
@ -74,7 +77,8 @@ class ShowVersions(command.Lister):
version_data = session.get_all_version_data(
interface=interface,
region_name=parsed_args.region_name,
service_type=parsed_args.service)
service_type=parsed_args.service,
)
columns = [
"Region Name",
@ -97,13 +101,15 @@ class ShowVersions(command.Lister):
for data in service_versions:
if status and status != data['status']:
continue
versions.append((
region_name,
service_type,
data['version'],
data['status'],
data['url'],
data['min_microversion'],
data['max_microversion'],
))
versions.append(
(
region_name,
service_type,
data['version'],
data['status'],
data['url'],
data['min_microversion'],
data['max_microversion'],
)
)
return (columns, versions)

12
openstackclient/tests/functional/common/test_availability_zone.py
View File

@ -14,7 +14,7 @@ from openstackclient.tests.functional import base
class AvailabilityZoneTests(base.TestCase):
"""Functional tests for availability zone. """
"""Functional tests for availability zone."""
def test_availability_zone_list(self):
cmd_output = self.openstack(
@ -22,11 +22,5 @@ class AvailabilityZoneTests(base.TestCase):
parse_output=True,
)
zones = [x['Zone Name'] for x in cmd_output]
self.assertIn(
'internal',
zones
)
self.assertIn(
'nova',
zones
)
self.assertIn('internal', zones)
self.assertIn('nova', zones)

26
openstackclient/tests/functional/common/test_configuration.py
View File

@ -23,17 +23,13 @@ class ConfigurationTests(base.TestCase):
"""Functional test for configuration."""
def test_configuration_show(self):
# Test show without option
raw_output = self.openstack('configuration show')
items = self.parse_listing(raw_output)
self.assert_table_structure(items, BASIC_CONFIG_HEADERS)
cmd_output = self.openstack('configuration show', parse_output=True)
self.assertEqual(
configuration.REDACTED,
cmd_output['auth.password']
)
self.assertEqual(configuration.REDACTED, cmd_output['auth.password'])
self.assertIn(
'auth.password',
cmd_output.keys(),
@ -41,30 +37,25 @@ class ConfigurationTests(base.TestCase):
# Test show --mask
cmd_output = self.openstack(
'configuration show --mask', parse_output=True,
)
self.assertEqual(
configuration.REDACTED,
cmd_output['auth.password']
'configuration show --mask',
parse_output=True,
)
self.assertEqual(configuration.REDACTED, cmd_output['auth.password'])
# Test show --unmask
cmd_output = self.openstack(
'configuration show --unmask', parse_output=True,
'configuration show --unmask',
parse_output=True,
)
# If we are using os-client-config, this will not be set. Rather than
# parse clouds.yaml to get the right value, just make sure
# we are not getting redacted.
passwd = os.environ.get('OS_PASSWORD')
if passwd:
self.assertEqual(
passwd,
cmd_output['auth.password']
)
self.assertEqual(passwd, cmd_output['auth.password'])
else:
self.assertNotEqual(
configuration.REDACTED,
cmd_output['auth.password']
configuration.REDACTED, cmd_output['auth.password']
)
@ -72,7 +63,6 @@ class ConfigurationTestsNoAuth(base.TestCase):
"""Functional test for configuration with no auth"""
def test_configuration_show(self):
# Test show without option
raw_output = self.openstack(
'configuration show',

16
openstackclient/tests/functional/common/test_help.py
View File

@ -27,8 +27,10 @@ class HelpTests(base.TestCase):
('server create', 'Create a new server'),
('server delete', 'Delete server(s)'),
('server dump create', 'Create a dump file in server(s)'),
('server image create',
'Create a new server disk image from an existing server'),
(
'server image create',
'Create a new server disk image from an existing server',
),
('server list', 'List servers'),
('server lock', 'Lock server(s)'),
('server migrate', 'Migrate server to different host'),
@ -51,7 +53,7 @@ class HelpTests(base.TestCase):
('server unpause', 'Unpause server(s)'),
('server unrescue', 'Restore server from rescue mode'),
('server unset', 'Unset server properties'),
('server unshelve', 'Unshelve server(s)')
('server unshelve', 'Unshelve server(s)'),
]
def test_server_commands_main_help(self):
@ -59,10 +61,14 @@ class HelpTests(base.TestCase):
raw_output = self.openstack('help')
for command, description in self.SERVER_COMMANDS:
msg = 'Command: %s not found in help output:\n%s' % (
command, raw_output)
command,
raw_output,
)
self.assertIn(command, raw_output, msg)
msg = 'Description: %s not found in help output:\n%s' % (
description, raw_output)
description,
raw_output,
)
self.assertIn(description, raw_output, msg)
def test_server_only_help(self):

17
openstackclient/tests/functional/common/test_module.py
View File

@ -18,13 +18,9 @@ from openstackclient.tests.functional import base
class ModuleTest(base.TestCase):
"""Functional tests for openstackclient module list output."""
CLIENTS = ['openstackclient',
'keystoneclient',
'novaclient',
'openstack']
CLIENTS = ['openstackclient', 'keystoneclient', 'novaclient', 'openstack']
LIBS = ['osc_lib',
'keystoneauth1']
LIBS = ['osc_lib', 'keystoneauth1']
def test_module_list(self):
# Test module list
@ -42,6 +38,7 @@ class ModuleTest(base.TestCase):
class CommandTest(base.TestCase):
"""Functional tests for openstackclient command list."""
GROUPS = [
'openstack.volume.v3',
'openstack.network.v2',
@ -59,13 +56,7 @@ class CommandTest(base.TestCase):
self.assertIn(one_group, group_names)
def test_command_list_with_group(self):
input_groups = [
'volume',
'network',
'image',
'identity',
'compute.v2'
]
input_groups = ['volume', 'network', 'image', 'identity', 'compute.v2']
for each_input in input_groups:
cmd_output = self.openstack(
'command list --group %s' % each_input,

50
openstackclient/tests/functional/common/test_quota.py
View File

@ -30,8 +30,9 @@ class QuotaTests(base.TestCase):
def setUpClass(cls):
super(QuotaTests, cls).setUpClass()
cls.haz_network = cls.is_service_enabled('network')
cls.PROJECT_NAME =\
cls.get_openstack_configuration_value('auth.project_name')
cls.PROJECT_NAME = cls.get_openstack_configuration_value(
'auth.project_name'
)
def test_quota_list_details_compute(self):
expected_headers = ["Resource", "In Use", "Reserved", "Limit"]
@ -111,9 +112,9 @@ class QuotaTests(base.TestCase):
if self.haz_network:
network_option = "--routers 21 "
self.openstack(
'quota set --cores 31 --backups 41 ' +
network_option +
self.PROJECT_NAME
'quota set --cores 31 --backups 41 '
+ network_option
+ self.PROJECT_NAME
)
cmd_output = self.openstack(
'quota show ' + self.PROJECT_NAME,
@ -151,8 +152,7 @@ class QuotaTests(base.TestCase):
def test_quota_set_class(self):
self.openstack(
'quota set --key-pairs 33 --snapshots 43 ' +
'--class default'
'quota set --key-pairs 33 --snapshots 43 ' + '--class default'
)
cmd_output = self.openstack(
'quota show --class default',
@ -194,8 +194,12 @@ class QuotaTests(base.TestCase):
'quota list --network',
parse_output=True,
)
self.addCleanup(self._restore_quota_limit, 'network',
cmd_output[0]['Networks'], self.PROJECT_NAME)
self.addCleanup(
self._restore_quota_limit,
'network',
cmd_output[0]['Networks'],
self.PROJECT_NAME,
)
self.openstack('quota set --networks 40 ' + self.PROJECT_NAME)
cmd_output = self.openstack(
@ -207,12 +211,16 @@ class QuotaTests(base.TestCase):
# That will ensure we have at least two networks in the system.
for _ in range(2):
self.openstack('network create --project %s %s' %
(self.PROJECT_NAME, uuid.uuid4().hex))
self.openstack(
'network create --project %s %s'
% (self.PROJECT_NAME, uuid.uuid4().hex)
)
self.assertRaises(exceptions.CommandFailed, self.openstack,
'quota set --networks 1 --no-force ' +
self.PROJECT_NAME)
self.assertRaises(
exceptions.CommandFailed,
self.openstack,
'quota set --networks 1 --no-force ' + self.PROJECT_NAME,
)
def test_quota_network_set_with_force(self):
self.skipTest('story 2010110')
@ -232,8 +240,12 @@ class QuotaTests(base.TestCase):
'quota list --network',
parse_output=True,
)
self.addCleanup(self._restore_quota_limit, 'network',
cmd_output[0]['Networks'], self.PROJECT_NAME)
self.addCleanup(
self._restore_quota_limit,
'network',
cmd_output[0]['Networks'],
self.PROJECT_NAME,
)
self.openstack('quota set --networks 40 ' + self.PROJECT_NAME)
cmd_output = self.openstack(
@ -245,8 +257,10 @@ class QuotaTests(base.TestCase):
# That will ensure we have at least two networks in the system.
for _ in range(2):
self.openstack('network create --project %s %s' %
(self.PROJECT_NAME, uuid.uuid4().hex))
self.openstack(
'network create --project %s %s'
% (self.PROJECT_NAME, uuid.uuid4().hex)
)
self.openstack('quota set --networks 1 --force ' + self.PROJECT_NAME)
cmd_output = self.openstack(

29
openstackclient/tests/unit/common/test_availability_zone.py
View File

@ -53,7 +53,10 @@ def _build_volume_az_datalist(volume_az, long_datalist=False):
datalist = (
volume_az.zoneName,
'available',
'', '', '', '',
'',
'',
'',
'',
)
return (datalist,)
@ -70,13 +73,14 @@ def _build_network_az_datalist(network_az, long_datalist=False):
network_az.name,
network_az.state,
network_az.resource,
'', '', '',
'',
'',
'',
)
return (datalist,)
class TestAvailabilityZone(utils.TestCommand):
def setUp(self):
super().setUp()
@ -110,9 +114,9 @@ class TestAvailabilityZone(utils.TestCommand):
class TestAvailabilityZoneList(TestAvailabilityZone):
compute_azs = \
compute_azs = (
compute_fakes.FakeAvailabilityZone.create_availability_zones()
)
volume_azs = volume_fakes.create_availability_zones(count=1)
network_azs = network_fakes.create_availability_zones()
@ -181,14 +185,17 @@ class TestAvailabilityZoneList(TestAvailabilityZone):
self.assertEqual(self.long_columnslist, columns)
datalist = ()
for compute_az in self.compute_azs:
datalist += _build_compute_az_datalist(compute_az,
long_datalist=True)
datalist += _build_compute_az_datalist(
compute_az, long_datalist=True
)
for volume_az in self.volume_azs:
datalist += _build_volume_az_datalist(volume_az,
long_datalist=True)
datalist += _build_volume_az_datalist(
volume_az, long_datalist=True
)
for network_az in self.network_azs:
datalist += _build_network_az_datalist(network_az,
long_datalist=True)
datalist += _build_network_az_datalist(
network_az, long_datalist=True
)
self.assertEqual(datalist, tuple(data))
def test_availability_zone_list_compute(self):

11
openstackclient/tests/unit/common/test_clientmanager.py
View File

@ -23,7 +23,6 @@ from openstackclient.tests.unit import fakes
class TestClientManager(osc_lib_test_utils.TestClientManager):
def _clientmanager_class(self):
"""Allow subclasses to override the ClientManager class"""
return clientmanager.ClientManager
@ -54,10 +53,12 @@ class TestClientManager(osc_lib_test_utils.TestClientManager):
def test_client_manager_network_endpoint_disabled(self):
auth_args = copy.deepcopy(self.default_password_auth)
auth_args.update({
'user_domain_name': 'default',
'project_domain_name': 'default',
})
auth_args.update(
{
'user_domain_name': 'default',
'project_domain_name': 'default',
}
)
# v3 fake doesn't have network endpoint
client_manager = self._make_clientmanager(
auth_args=auth_args,

13
openstackclient/tests/unit/common/test_command.py
View File

@ -22,18 +22,18 @@ from openstackclient.tests.unit import utils as test_utils
class FakeCommand(command.Command):
def take_action(self, parsed_args):
pass
class TestCommand(test_utils.TestCase):
def test_command_has_logger(self):
cmd = FakeCommand(mock.Mock(), mock.Mock())
self.assertTrue(hasattr(cmd, 'log'))
self.assertEqual('openstackclient.tests.unit.common.test_command.'
'FakeCommand', cmd.log.name)
self.assertEqual(
'openstackclient.tests.unit.common.test_command.' 'FakeCommand',
cmd.log.name,
)
def test_validate_os_beta_command_enabled(self):
cmd = FakeCommand(mock.Mock(), mock.Mock())
@ -45,5 +45,6 @@ class TestCommand(test_utils.TestCase):
cmd.validate_os_beta_command_enabled()
cmd.app.options.os_beta_command = False
self.assertRaises(exceptions.CommandError,
cmd.validate_os_beta_command_enabled)
self.assertRaises(
exceptions.CommandError, cmd.validate_os_beta_command_enabled
)

1
openstackclient/tests/unit/common/test_configuration.py
View File

@ -19,7 +19,6 @@ from openstackclient.tests.unit import utils
class TestConfiguration(utils.TestCommand):
columns = (
'auth.password',
'auth.token',

103
openstackclient/tests/unit/common/test_extension.py
View File

@ -24,7 +24,6 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
class TestExtension(utils.TestCommand):
def setUp(self):
super().setUp()
@ -61,10 +60,15 @@ class TestExtension(utils.TestCommand):
class TestExtensionList(TestExtension):
columns = ('Name', 'Alias', 'Description')
long_columns = ('Name', 'Alias', 'Description', 'Namespace', 'Updated',
'Links')
long_columns = (
'Name',
'Alias',
'Description',
'Namespace',
'Updated',
'Links',
)
volume_extension = volume_fakes.create_one_extension()
identity_extension = identity_fakes.FakeExtension.create_one_extension()
@ -75,17 +79,20 @@ class TestExtensionList(TestExtension):
super().setUp()
self.identity_extensions_mock.list.return_value = [
self.identity_extension]
self.identity_extension
]
self.compute_extensions_mock.return_value = [self.compute_extension]
self.volume_extensions_mock.show_all.return_value = [
self.volume_extension]
self.volume_extension
]
self.network_extensions_mock.return_value = [self.network_extension]
# Get the command object to test
self.cmd = extension.ListExtension(self.app, None)
def _test_extension_list_helper(self, arglist, verifylist,
expected_data, long=False):
def _test_extension_list_helper(
self, arglist, verifylist, expected_data, long=False
):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# In base command class Lister in cliff, abstract method take_action()
@ -184,11 +191,13 @@ class TestExtensionList(TestExtension):
verifylist = [
('identity', True),
]
datalist = ((
self.identity_extension.name,
self.identity_extension.alias,
self.identity_extension.description,
), )
datalist = (
(
self.identity_extension.name,
self.identity_extension.alias,
self.identity_extension.description,
),
)
self._test_extension_list_helper(arglist, verifylist, datalist)
self.identity_extensions_mock.list.assert_called_with()
@ -218,16 +227,19 @@ class TestExtensionList(TestExtension):
('network', True),
('long', True),
]
datalist = ((
self.network_extension.name,
self.network_extension.alias,
self.network_extension.description,
self.network_extension.namespace,
self.network_extension.updated,
self.network_extension.links,
), )
self._test_extension_list_helper(arglist, verifylist, datalist,
long=True)
datalist = (
(
self.network_extension.name,
self.network_extension.alias,
self.network_extension.description,
self.network_extension.namespace,
self.network_extension.updated,
self.network_extension.links,
),
)
self._test_extension_list_helper(
arglist, verifylist, datalist, long=True
)
self.network_extensions_mock.assert_called_with()
def test_extension_list_compute(self):
@ -237,11 +249,13 @@ class TestExtensionList(TestExtension):
verifylist = [
('compute', True),
]
datalist = ((
self.compute_extension.name,
self.compute_extension.alias,
self.compute_extension.description,
), )
datalist = (
(
self.compute_extension.name,
self.compute_extension.alias,
self.compute_extension.description,
),
)
self._test_extension_list_helper(arglist, verifylist, datalist)
self.compute_extensions_mock.assert_called_with()
@ -277,19 +291,19 @@ class TestExtensionList(TestExtension):
verifylist = [
('volume', True),
]
datalist = ((
self.volume_extension.name,
self.volume_extension.alias,
self.volume_extension.description,
), )
datalist = (
(
self.volume_extension.name,
self.volume_extension.alias,
self.volume_extension.description,
),
)
self._test_extension_list_helper(arglist, verifylist, datalist)
self.volume_extensions_mock.show_all.assert_called_with()
class TestExtensionShow(TestExtension):
extension_details = (
network_fakes.FakeExtension.create_one_extension()
)
extension_details = network_fakes.FakeExtension.create_one_extension()
columns = (
'alias',
@ -306,7 +320,7 @@ class TestExtensionShow(TestExtension):
extension_details.links,
extension_details.name,
extension_details.namespace,
extension_details.updated
extension_details.updated,
)
def setUp(self):
@ -315,14 +329,20 @@ class TestExtensionShow(TestExtension):
self.cmd = extension.ShowExtension(self.app, None)
self.app.client_manager.network.find_extension = mock.Mock(
return_value=self.extension_details)
return_value=self.extension_details
)
def test_show_no_options(self):
arglist = []
verifylist = []
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
self.assertRaises(
tests_utils.ParserException,
self.check_parser,
self.cmd,
arglist,
verifylist,
)
def test_show_all_options(self):
arglist = [
@ -337,7 +357,8 @@ class TestExtensionShow(TestExtension):
columns, data = self.cmd.take_action(parsed_args)
self.app.client_manager.network.find_extension.assert_called_with(
self.extension_details.alias, ignore_missing=False)
self.extension_details.alias, ignore_missing=False
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)

19
openstackclient/tests/unit/common/test_limits.py
View File

@ -17,20 +17,12 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
class TestComputeLimits(compute_fakes.TestComputev2):
absolute_columns = [
'Name',
'Value',
]
rate_columns = [
"Verb",
"URI",
"Value",
"Remain",
"Unit",
"Next Available"
]
rate_columns = ["Verb", "URI", "Value", "Remain", "Unit", "Next Available"]
def setUp(self):
super().setUp()
@ -77,14 +69,7 @@ class TestVolumeLimits(volume_fakes.TestVolume):
'Value',
]
rate_columns = [
"Verb",
"URI",
"Value",
"Remain",
"Unit",
"Next Available"
]
rate_columns = ["Verb", "URI", "Value", "Remain", "Unit", "Next Available"]
def setUp(self):
super().setUp()

39
openstackclient/tests/unit/common/test_logs.py
View File

@ -23,7 +23,6 @@ from openstackclient.tests.unit import utils
class TestContext(utils.TestCase):
def test_log_level_from_options(self):
opts = mock.Mock()
opts.verbose_level = 0
@ -70,35 +69,47 @@ class TestContext(utils.TestCase):
class TestFileFormatter(utils.TestCase):
def test_nothing(self):
formatter = logs._FileFormatter()
self.assertEqual(('%(asctime)s.%(msecs)03d %(process)d %(levelname)s '
'%(name)s %(message)s'), formatter.fmt)
self.assertEqual(
(
'%(asctime)s.%(msecs)03d %(process)d %(levelname)s '
'%(name)s %(message)s'
),
formatter.fmt,
)
def test_options(self):
class Opts(object):
cloud = 'cloudy'
os_project_name = 'projecty'
username = 'usernamey'
options = Opts()
formatter = logs._FileFormatter(options=options)
self.assertEqual(('%(asctime)s.%(msecs)03d %(process)d %(levelname)s '
'%(name)s [cloudy usernamey projecty] %(message)s'),
formatter.fmt)
self.assertEqual(
(
'%(asctime)s.%(msecs)03d %(process)d %(levelname)s '
'%(name)s [cloudy usernamey projecty] %(message)s'
),
formatter.fmt,
)
def test_config(self):
config = mock.Mock()
config.config = {'cloud': 'cloudy'}
config.auth = {'project_name': 'projecty', 'username': 'usernamey'}
formatter = logs._FileFormatter(config=config)
self.assertEqual(('%(asctime)s.%(msecs)03d %(process)d %(levelname)s '
'%(name)s [cloudy usernamey projecty] %(message)s'),
formatter.fmt)
self.assertEqual(
(
'%(asctime)s.%(msecs)03d %(process)d %(levelname)s '
'%(name)s [cloudy usernamey projecty] %(message)s'
),
formatter.fmt,
)
class TestLogConfigurator(utils.TestCase):
def setUp(self):
super(TestLogConfigurator, self).setUp()
self.options = mock.Mock()
@ -121,7 +132,8 @@ class TestLogConfigurator(utils.TestCase):
self.requests_log,
self.cliff_log,
self.stevedore_log,
self.iso8601_log]
self.iso8601_log,
]
@mock.patch('logging.StreamHandler')
@mock.patch('logging.getLogger')
@ -190,7 +202,8 @@ class TestLogConfigurator(utils.TestCase):
cloud_config.config = {
'log_file': config_log,
'verbose_level': 1,
'log_level': 'info'}
'log_level': 'info',
}
file_logger = mock.Mock()
file_logger.setFormatter = mock.Mock()
file_logger.setLevel = mock.Mock()

18
openstackclient/tests/unit/common/test_module.py
View File

@ -52,7 +52,6 @@ MODULES = {
class TestCommandList(utils.TestCommand):
def setUp(self):
super(TestCommandList, self).setUp()
@ -81,16 +80,14 @@ class TestCommandList(utils.TestCommand):
# handling the detection rather than using the hard-code below.
collist = ('Command Group', 'Commands')
self.assertEqual(collist, columns)
datalist = ((
'openstack.common',
'limits show\nextension list'
),)
datalist = (('openstack.common', 'limits show\nextension list'),)
self.assertEqual(datalist, tuple(data))
def test_command_list_with_group_not_found(self):
arglist = [
'--group', 'not_exist',
'--group',
'not_exist',
]
verifylist = [
('group', 'not_exist'),
@ -105,7 +102,8 @@ class TestCommandList(utils.TestCommand):
def test_command_list_with_group(self):
arglist = [
'--group', 'common',
'--group',
'common',
]
verifylist = [
('group', 'common'),
@ -116,10 +114,7 @@ class TestCommandList(utils.TestCommand):
collist = ('Command Group', 'Commands')
self.assertEqual(collist, columns)
datalist = ((
'openstack.common',
'limits show\nextension list'
),)
datalist = (('openstack.common', 'limits show\nextension list'),)
self.assertEqual(datalist, tuple(data))
@ -130,7 +125,6 @@ class TestCommandList(utils.TestCommand):
clear=True,
)
class TestModuleList(utils.TestCommand):
def setUp(self):
super(TestModuleList, self).setUp()

87
openstackclient/tests/unit/common/test_parseractions.py
View File

@ -24,7 +24,6 @@ from openstackclient.tests.unit import utils
class TestKeyValueAction(utils.TestCase):
def setUp(self):
super(TestKeyValueAction, self).setUp()
@ -37,15 +36,20 @@ class TestKeyValueAction(utils.TestCase):
action=parseractions.KeyValueAction,
default={'green': '20%', 'format': '#rgb'},
help='Property to store for this volume '
'(repeat option to set multiple properties)',
'(repeat option to set multiple properties)',
)
def test_good_values(self):
results = self.parser.parse_args([
'--property', 'red=',
'--property', 'green=100%',
'--property', 'blue=50%',
])
results = self.parser.parse_args(
[
'--property',
'red=',
'--property',
'green=100%',
'--property',
'blue=50%',
]
)
actual = getattr(results, 'property', {})
# All should pass through unmolested
@ -57,13 +61,13 @@ class TestKeyValueAction(utils.TestCase):
argparse.ArgumentTypeError,
self.parser.parse_args,
[
'--property', 'red',
]
'--property',
'red',
],
)
class TestMultiKeyValueAction(utils.TestCase):
def setUp(self):
super(TestMultiKeyValueAction, self).setUp()
@ -78,14 +82,18 @@ class TestMultiKeyValueAction(utils.TestCase):
default=None,
required_keys=['req1', 'req2'],
optional_keys=['opt1', 'opt2'],
help='Test'
help='Test',
)
def test_good_values(self):
results = self.parser.parse_args([
'--test', 'req1=aaa,req2=bbb',
'--test', 'req1=,req2=',
])
results = self.parser.parse_args(
[
'--test',
'req1=aaa,req2=bbb',
'--test',
'req1=,req2=',
]
)
actual = getattr(results, 'test', [])
expect = [
@ -103,13 +111,17 @@ class TestMultiKeyValueAction(utils.TestCase):
default=None,
required_keys=[],
optional_keys=[],
help='Test'
help='Test',
)
results = self.parser.parse_args([
'--test-empty', 'req1=aaa,req2=bbb',
'--test-empty', 'req1=,req2=',
])
results = self.parser.parse_args(
[
'--test-empty',
'req1=aaa,req2=bbb',
'--test-empty',
'req1=,req2=',
]
)
actual = getattr(results, 'test_empty', [])
expect = [
@ -123,8 +135,9 @@ class TestMultiKeyValueAction(utils.TestCase):
argparse.ArgumentTypeError,
self.parser.parse_args,
[
'--test', 'mmm,nnn=zzz',
]
'--test',
'mmm,nnn=zzz',
],
)
def test_error_values_without_comma(self):
@ -132,8 +145,9 @@ class TestMultiKeyValueAction(utils.TestCase):
argparse.ArgumentTypeError,
self.parser.parse_args,
[
'--test', 'mmmnnn',
]
'--test',
'mmmnnn',
],
)
def test_missing_key(self):
@ -141,8 +155,9 @@ class TestMultiKeyValueAction(utils.TestCase):
argparse.ArgumentTypeError,
self.parser.parse_args,
[
'--test', 'req2=ddd',
]
'--test',
'req2=ddd',
],
)
def test_invalid_key(self):
@ -150,8 +165,9 @@ class TestMultiKeyValueAction(utils.TestCase):
argparse.ArgumentTypeError,
self.parser.parse_args,
[
'--test', 'req1=aaa,req2=bbb,aaa=req1',
]
'--test',
'req1=aaa,req2=bbb,aaa=req1',
],
)
def test_required_keys_not_list(self):
@ -165,7 +181,7 @@ class TestMultiKeyValueAction(utils.TestCase):
default=None,
required_keys={'aaa': 'bbb'},
optional_keys=['opt1', 'opt2'],
help='Test'
help='Test',
)
def test_optional_keys_not_list(self):
@ -179,12 +195,11 @@ class TestMultiKeyValueAction(utils.TestCase):
default=None,
required_keys=['req1', 'req2'],
optional_keys={'aaa': 'bbb'},
help='Test'
help='Test',
)
class TestNonNegativeAction(utils.TestCase):
def setUp(self):
super(TestNonNegativeAction, self).setUp()
@ -202,21 +217,17 @@ class TestNonNegativeAction(utils.TestCase):
self.assertRaises(
argparse.ArgumentTypeError,
self.parser.parse_args,
"--foo -1".split()
"--foo -1".split(),
)
def test_zero_values(self):
results = self.parser.parse_args(
'--foo 0'.split()
)
results = self.parser.parse_args('--foo 0'.split())
actual = getattr(results, 'foo', None)
self.assertEqual(actual, 0)
def test_positive_values(self):
results = self.parser.parse_args(
'--foo 1'.split()
)
results = self.parser.parse_args('--foo 1'.split())
actual = getattr(results, 'foo', None)
self.assertEqual(actual, 1)

6
openstackclient/tests/unit/common/test_progressbar.py
View File

@ -19,7 +19,6 @@ from openstackclient.tests.unit import utils
class TestProgressBarWrapper(utils.TestCase):
def test_iter_file_display_progress_bar(self):
size = 98304
file_obj = io.StringIO('X' * size)
@ -31,10 +30,7 @@ class TestProgressBarWrapper(utils.TestCase):
chunk = file_obj.read(chunksize)
while chunk:
chunk = file_obj.read(chunksize)
self.assertEqual(
'[%s>] 100%%\n' % ('=' * 29),
output.getvalue()
)
self.assertEqual('[%s>] 100%%\n' % ('=' * 29), output.getvalue())
finally:
sys.stdout = saved_stdout

79
openstackclient/tests/unit/common/test_project_cleanup.py
View File

@ -19,7 +19,6 @@ from openstackclient.tests.unit import utils as tests_utils
class TestProjectCleanupBase(tests_utils.TestCommand):
def setUp(self):
super(TestProjectCleanupBase, self).setUp()
@ -27,7 +26,6 @@ class TestProjectCleanupBase(tests_utils.TestCommand):
class TestProjectCleanup(TestProjectCleanupBase):
project = identity_fakes.FakeProject.create_one_project()
def setUp(self):
@ -35,34 +33,46 @@ class TestProjectCleanup(TestProjectCleanupBase):
self.cmd = project_cleanup.ProjectCleanup(self.app, None)
self.project_cleanup_mock = mock.Mock()
self.sdk_connect_as_project_mock = \
mock.Mock(return_value=self.app.client_manager.sdk_connection)
self.app.client_manager.sdk_connection.project_cleanup = \
self.sdk_connect_as_project_mock = mock.Mock(
return_value=self.app.client_manager.sdk_connection
)
self.app.client_manager.sdk_connection.project_cleanup = (
self.project_cleanup_mock
self.app.client_manager.sdk_connection.identity.find_project = \
)
self.app.client_manager.sdk_connection.identity.find_project = (
mock.Mock(return_value=self.project)
self.app.client_manager.sdk_connection.connect_as_project = \
)
self.app.client_manager.sdk_connection.connect_as_project = (
self.sdk_connect_as_project_mock
)
def test_project_no_options(self):
arglist = []
verifylist = []
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
self.assertRaises(
tests_utils.ParserException,
self.check_parser,
self.cmd,
arglist,
verifylist,
)
def test_project_cleanup_with_filters(self):
arglist = [
'--project', self.project.id,
'--created-before', '2200-01-01',
'--updated-before', '2200-01-02'
'--project',
self.project.id,
'--created-before',
'2200-01-01',
'--updated-before',
'2200-01-02',
]
verifylist = [
('dry_run', False),
('auth_project', False),
('project', self.project.id),
('created_before', '2200-01-01'),
('updated_before', '2200-01-02')
('updated_before', '2200-01-02'),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = None
@ -70,16 +80,12 @@ class TestProjectCleanup(TestProjectCleanupBase):
with mock.patch('sys.stdin', StringIO('y')):
result = self.cmd.take_action(parsed_args)
self.sdk_connect_as_project_mock.assert_called_with(
self.project)
filters = {
'created_at': '2200-01-01',
'updated_at': '2200-01-02'
}
self.sdk_connect_as_project_mock.assert_called_with(self.project)
filters = {'created_at': '2200-01-01', 'updated_at': '2200-01-02'}
calls = [
mock.call(dry_run=True, status_queue=mock.ANY, filters=filters),
mock.call(dry_run=False, status_queue=mock.ANY, filters=filters)
mock.call(dry_run=False, status_queue=mock.ANY, filters=filters),
]
self.project_cleanup_mock.assert_has_calls(calls)
@ -87,7 +93,8 @@ class TestProjectCleanup(TestProjectCleanupBase):
def test_project_cleanup_with_auto_approve(self):
arglist = [
'--project', self.project.id,
'--project',
self.project.id,
'--auto-approve',
]
verifylist = [
@ -101,11 +108,10 @@ class TestProjectCleanup(TestProjectCleanupBase):
result = self.cmd.take_action(parsed_args)
self.sdk_connect_as_project_mock.assert_called_with(
self.project)
self.sdk_connect_as_project_mock.assert_called_with(self.project)
calls = [
mock.call(dry_run=True, status_queue=mock.ANY, filters={}),
mock.call(dry_run=False, status_queue=mock.ANY, filters={})
mock.call(dry_run=False, status_queue=mock.ANY, filters={}),
]
self.project_cleanup_mock.assert_has_calls(calls)
@ -113,7 +119,8 @@ class TestProjectCleanup(TestProjectCleanupBase):
def test_project_cleanup_with_project(self):
arglist = [
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', False),
@ -126,11 +133,10 @@ class TestProjectCleanup(TestProjectCleanupBase):
with mock.patch('sys.stdin', StringIO('y')):
result = self.cmd.take_action(parsed_args)
self.sdk_connect_as_project_mock.assert_called_with(
self.project)
self.sdk_connect_as_project_mock.assert_called_with(self.project)
calls = [
mock.call(dry_run=True, status_queue=mock.ANY, filters={}),
mock.call(dry_run=False, status_queue=mock.ANY, filters={})
mock.call(dry_run=False, status_queue=mock.ANY, filters={}),
]
self.project_cleanup_mock.assert_has_calls(calls)
@ -138,7 +144,8 @@ class TestProjectCleanup(TestProjectCleanupBase):
def test_project_cleanup_with_project_abort(self):
arglist = [
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', False),
@ -151,8 +158,7 @@ class TestProjectCleanup(TestProjectCleanupBase):
with mock.patch('sys.stdin', StringIO('n')):
result = self.cmd.take_action(parsed_args)
self.sdk_connect_as_project_mock.assert_called_with(
self.project)
self.sdk_connect_as_project_mock.assert_called_with(self.project)
calls = [
mock.call(dry_run=True, status_queue=mock.ANY, filters={}),
]
@ -163,7 +169,8 @@ class TestProjectCleanup(TestProjectCleanupBase):
def test_project_cleanup_with_dry_run(self):
arglist = [
'--dry-run',
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', True),
@ -175,10 +182,10 @@ class TestProjectCleanup(TestProjectCleanupBase):
result = self.cmd.take_action(parsed_args)
self.sdk_connect_as_project_mock.assert_called_with(
self.project)
self.sdk_connect_as_project_mock.assert_called_with(self.project)
self.project_cleanup_mock.assert_called_once_with(
dry_run=True, status_queue=mock.ANY, filters={})
dry_run=True, status_queue=mock.ANY, filters={}
)
self.assertIsNone(result)
@ -202,7 +209,7 @@ class TestProjectCleanup(TestProjectCleanupBase):
self.sdk_connect_as_project_mock.assert_not_called()
calls = [
mock.call(dry_run=True, status_queue=mock.ANY, filters={}),
mock.call(dry_run=False, status_queue=mock.ANY, filters={})
mock.call(dry_run=False, status_queue=mock.ANY, filters={}),
]
self.project_cleanup_mock.assert_has_calls(calls)

134
openstackclient/tests/unit/common/test_project_purge.py
View File

@ -24,7 +24,6 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
class TestProjectPurgeInit(tests_utils.TestCommand):
def setUp(self):
super().setUp()
compute_client = compute_fakes.FakeComputev2Client(
@ -67,7 +66,6 @@ class TestProjectPurgeInit(tests_utils.TestCommand):
class TestProjectPurge(TestProjectPurgeInit):
project = identity_fakes.FakeProject.create_one_project()
server = compute_fakes.FakeServer.create_one_server()
image = image_fakes.create_one_image()
@ -97,12 +95,18 @@ class TestProjectPurge(TestProjectPurgeInit):
arglist = []
verifylist = []
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
self.assertRaises(
tests_utils.ParserException,
self.check_parser,
self.cmd,
arglist,
verifylist,
)
def test_project_purge_with_project(self):
arglist = [
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', False),
@ -117,17 +121,23 @@ class TestProjectPurge(TestProjectPurgeInit):
self.projects_mock.get.assert_called_once_with(self.project.id)
self.projects_mock.delete.assert_called_once_with(self.project.id)
self.servers_mock.list.assert_called_once_with(
search_opts={'tenant_id': self.project.id, 'all_tenants': True})
search_opts={'tenant_id': self.project.id, 'all_tenants': True}
)
kwargs = {'filters': {'owner': self.project.id}}
self.images_mock.list.assert_called_once_with(**kwargs)
volume_search_opts = {'project_id': self.project.id,
'all_tenants': True}
volume_search_opts = {
'project_id': self.project.id,
'all_tenants': True,
}
self.volumes_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.snapshots_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.backups_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.servers_mock.delete.assert_called_once_with(self.server.id)
self.images_mock.delete.assert_called_once_with(self.image.id)
self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
@ -138,7 +148,8 @@ class TestProjectPurge(TestProjectPurgeInit):
def test_project_purge_with_dry_run(self):
arglist = [
'--dry-run',
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', True),
@ -153,17 +164,23 @@ class TestProjectPurge(TestProjectPurgeInit):
self.projects_mock.get.assert_called_once_with(self.project.id)
self.projects_mock.delete.assert_not_called()
self.servers_mock.list.assert_called_once_with(
search_opts={'tenant_id': self.project.id, 'all_tenants': True})
search_opts={'tenant_id': self.project.id, 'all_tenants': True}
)
kwargs = {'filters': {'owner': self.project.id}}
self.images_mock.list.assert_called_once_with(**kwargs)
volume_search_opts = {'project_id': self.project.id,
'all_tenants': True}
volume_search_opts = {
'project_id': self.project.id,
'all_tenants': True,
}
self.volumes_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.snapshots_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.backups_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.servers_mock.delete.assert_not_called()
self.images_mock.delete.assert_not_called()
self.volumes_mock.force_delete.assert_not_called()
@ -174,7 +191,8 @@ class TestProjectPurge(TestProjectPurgeInit):
def test_project_purge_with_keep_project(self):
arglist = [
'--keep-project',
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', False),
@ -189,17 +207,23 @@ class TestProjectPurge(TestProjectPurgeInit):
self.projects_mock.get.assert_called_once_with(self.project.id)
self.projects_mock.delete.assert_not_called()
self.servers_mock.list.assert_called_once_with(
search_opts={'tenant_id': self.project.id, 'all_tenants': True})
search_opts={'tenant_id': self.project.id, 'all_tenants': True}
)
kwargs = {'filters': {'owner': self.project.id}}
self.images_mock.list.assert_called_once_with(**kwargs)
volume_search_opts = {'project_id': self.project.id,
'all_tenants': True}
volume_search_opts = {
'project_id': self.project.id,
'all_tenants': True,
}
self.volumes_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.snapshots_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.backups_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.servers_mock.delete.assert_called_once_with(self.server.id)
self.images_mock.delete.assert_called_once_with(self.image.id)
self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
@ -226,17 +250,23 @@ class TestProjectPurge(TestProjectPurgeInit):
self.projects_mock.get.assert_not_called()
self.projects_mock.delete.assert_called_once_with(self.project.id)
self.servers_mock.list.assert_called_once_with(
search_opts={'tenant_id': self.project.id, 'all_tenants': True})
search_opts={'tenant_id': self.project.id, 'all_tenants': True}
)
kwargs = {'filters': {'owner': self.project.id}}
self.images_mock.list.assert_called_once_with(**kwargs)
volume_search_opts = {'project_id': self.project.id,
'all_tenants': True}
volume_search_opts = {
'project_id': self.project.id,
'all_tenants': True,
}
self.volumes_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.snapshots_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.backups_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.servers_mock.delete.assert_called_once_with(self.server.id)
self.images_mock.delete.assert_called_once_with(self.image.id)
self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
@ -248,7 +278,8 @@ class TestProjectPurge(TestProjectPurgeInit):
def test_project_purge_with_exception(self, mock_error):
self.servers_mock.delete.side_effect = exceptions.CommandError()
arglist = [
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', False),
@ -263,17 +294,23 @@ class TestProjectPurge(TestProjectPurgeInit):
self.projects_mock.get.assert_called_once_with(self.project.id)
self.projects_mock.delete.assert_called_once_with(self.project.id)
self.servers_mock.list.assert_called_once_with(
search_opts={'tenant_id': self.project.id, 'all_tenants': True})
search_opts={'tenant_id': self.project.id, 'all_tenants': True}
)
kwargs = {'filters': {'owner': self.project.id}}
self.images_mock.list.assert_called_once_with(**kwargs)
volume_search_opts = {'project_id': self.project.id,
'all_tenants': True}
volume_search_opts = {
'project_id': self.project.id,
'all_tenants': True,
}
self.volumes_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.snapshots_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.backups_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.servers_mock.delete.assert_called_once_with(self.server.id)
self.images_mock.delete.assert_called_once_with(self.image.id)
self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
@ -285,7 +322,8 @@ class TestProjectPurge(TestProjectPurgeInit):
def test_project_purge_with_force_delete_backup(self):
self.backups_mock.delete.side_effect = [exceptions.CommandError, None]
arglist = [
'--project', self.project.id,
'--project',
self.project.id,
]
verifylist = [
('dry_run', False),
@ -300,17 +338,23 @@ class TestProjectPurge(TestProjectPurgeInit):
self.projects_mock.get.assert_called_once_with(self.project.id)
self.projects_mock.delete.assert_called_once_with(self.project.id)
self.servers_mock.list.assert_called_once_with(
search_opts={'tenant_id': self.project.id, 'all_tenants': True})
search_opts={'tenant_id': self.project.id, 'all_tenants': True}
)
kwargs = {'filters': {'owner': self.project.id}}
self.images_mock.list.assert_called_once_with(**kwargs)
volume_search_opts = {'project_id': self.project.id,
'all_tenants': True}
volume_search_opts = {
'project_id': self.project.id,
'all_tenants': True,
}
self.volumes_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.snapshots_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.backups_mock.list.assert_called_once_with(
search_opts=volume_search_opts)
search_opts=volume_search_opts
)
self.servers_mock.delete.assert_called_once_with(self.server.id)
self.images_mock.delete.assert_called_once_with(self.image.id)
self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)

321
openstackclient/tests/unit/common/test_quota.py
View File

@ -25,7 +25,6 @@ from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
class FakeQuotaResource(fakes.FakeResource):
_keys = {'property': 'value'}
def set_keys(self, args):
@ -40,7 +39,6 @@ class FakeQuotaResource(fakes.FakeResource):
class TestQuota(compute_fakes.TestComputev2):
def setUp(self):
super(TestQuota, self).setUp()
@ -52,14 +50,16 @@ class TestQuota(compute_fakes.TestComputev2):
self.compute_quotas_mock = self.app.client_manager.compute.quotas
self.compute_quotas_mock.reset_mock()
self.compute_quotas_class_mock = \
self.compute_quotas_class_mock = (
self.app.client_manager.compute.quota_classes
)
self.compute_quotas_class_mock.reset_mock()
self.volume_quotas_mock = self.app.client_manager.volume.quotas
self.volume_quotas_mock.reset_mock()
self.volume_quotas_class_mock = \
self.volume_quotas_class_mock = (
self.app.client_manager.volume.quota_classes
)
self.volume_quotas_class_mock.reset_mock()
self.app.client_manager.network = mock.Mock()
@ -67,8 +67,9 @@ class TestQuota(compute_fakes.TestComputev2):
self.app.client_manager.auth_ref = mock.Mock()
self.app.client_manager.auth_ref.service_catalog = mock.Mock()
self.service_catalog_mock = \
self.service_catalog_mock = (
self.app.client_manager.auth_ref.service_catalog
)
self.service_catalog_mock.reset_mock()
self.app.client_manager.auth_ref.project_id = identity_fakes.project_id
@ -101,7 +102,7 @@ class TestQuotaList(TestQuota):
'Security Groups',
'Security Group Rules',
'Subnets',
'Subnet Pools'
'Subnet Pools',
)
volume_column_header = (
@ -210,16 +211,14 @@ class TestQuotaList(TestQuota):
in_use = values['used']
else:
in_use = values['in_use']
resource_values = [
in_use,
values['reserved'],
values['limit']]
resource_values = [in_use, values['reserved'], values['limit']]
reference_data.append(tuple([name] + resource_values))
return reference_data
def test_quota_list_details_compute(self):
detailed_quota = (
compute_fakes.FakeQuota.create_one_comp_detailed_quota())
compute_fakes.FakeQuota.create_one_comp_detailed_quota()
)
detailed_column_header = (
'Resource',
@ -227,13 +226,15 @@ class TestQuotaList(TestQuota):
'Reserved',
'Limit',
)
detailed_reference_data = (
self._get_detailed_reference_data(detailed_quota))
detailed_reference_data = self._get_detailed_reference_data(
detailed_quota
)
self.compute.quotas.get = mock.Mock(return_value=detailed_quota)
arglist = [
'--detail', '--compute',
'--detail',
'--compute',
]
verifylist = [
('detail', True),
@ -245,12 +246,12 @@ class TestQuotaList(TestQuota):
ret_quotas = list(data)
self.assertEqual(detailed_column_header, columns)
self.assertEqual(
sorted(detailed_reference_data), sorted(ret_quotas))
self.assertEqual(sorted(detailed_reference_data), sorted(ret_quotas))
def test_quota_list_details_network(self):
detailed_quota = (
network_fakes.FakeQuota.create_one_net_detailed_quota())
network_fakes.FakeQuota.create_one_net_detailed_quota()
)
detailed_column_header = (
'Resource',
@ -258,13 +259,15 @@ class TestQuotaList(TestQuota):
'Reserved',
'Limit',
)
detailed_reference_data = (
self._get_detailed_reference_data(detailed_quota))
detailed_reference_data = self._get_detailed_reference_data(
detailed_quota
)
self.network.get_quota = mock.Mock(return_value=detailed_quota)
arglist = [
'--detail', '--network',
'--detail',
'--network',
]
verifylist = [
('detail', True),
@ -276,8 +279,7 @@ class TestQuotaList(TestQuota):
ret_quotas = list(data)
self.assertEqual(detailed_column_header, columns)
self.assertEqual(
sorted(detailed_reference_data), sorted(ret_quotas))
self.assertEqual(sorted(detailed_reference_data), sorted(ret_quotas))
def test_quota_list_details_volume(self):
detailed_quota = volume_fakes.create_one_detailed_quota()
@ -288,8 +290,9 @@ class TestQuotaList(TestQuota):
'Reserved',
'Limit',
)
detailed_reference_data = (
self._get_detailed_reference_data(detailed_quota))
detailed_reference_data = self._get_detailed_reference_data(
detailed_quota
)
self.volume.quotas.get = mock.Mock(return_value=detailed_quota)
@ -433,7 +436,8 @@ class TestQuotaList(TestQuota):
arglist = [
'--compute',
'--project', self.projects[0].name,
'--project',
self.projects[0].name,
]
verifylist = [
('compute', True),
@ -525,7 +529,8 @@ class TestQuotaList(TestQuota):
arglist = [
'--network',
'--project', self.projects[0].name,
'--project',
self.projects[0].name,
]
verifylist = [
('network', True),
@ -617,7 +622,8 @@ class TestQuotaList(TestQuota):
arglist = [
'--volume',
'--project', self.projects[0].name,
'--project',
self.projects[0].name,
]
verifylist = [
('volume', True),
@ -634,7 +640,6 @@ class TestQuotaList(TestQuota):
class TestQuotaSet(TestQuota):
def setUp(self):
super(TestQuotaSet, self).setUp()
@ -666,28 +671,44 @@ class TestQuotaSet(TestQuota):
def test_quota_set(self):
arglist = [
'--floating-ips', str(compute_fakes.floating_ip_num),
'--fixed-ips', str(compute_fakes.fix_ip_num),
'--injected-files', str(compute_fakes.injected_file_num),
'--injected-file-size', str(compute_fakes.injected_file_size_num),
'--injected-path-size', str(compute_fakes.injected_path_size_num),
'--key-pairs', str(compute_fakes.key_pair_num),
'--cores', str(compute_fakes.core_num),
'--ram', str(compute_fakes.ram_num),
'--instances', str(compute_fakes.instance_num),
'--properties', str(compute_fakes.property_num),
'--secgroup-rules', str(compute_fakes.secgroup_rule_num),
'--secgroups', str(compute_fakes.secgroup_num),
'--server-groups', str(compute_fakes.servgroup_num),
'--server-group-members', str(compute_fakes.servgroup_members_num),
'--floating-ips',
str(compute_fakes.floating_ip_num),
'--fixed-ips',
str(compute_fakes.fix_ip_num),
'--injected-files',
str(compute_fakes.injected_file_num),
'--injected-file-size',
str(compute_fakes.injected_file_size_num),
'--injected-path-size',
str(compute_fakes.injected_path_size_num),
'--key-pairs',
str(compute_fakes.key_pair_num),
'--cores',
str(compute_fakes.core_num),
'--ram',
str(compute_fakes.ram_num),
'--instances',
str(compute_fakes.instance_num),
'--properties',
str(compute_fakes.property_num),
'--secgroup-rules',
str(compute_fakes.secgroup_rule_num),
'--secgroups',
str(compute_fakes.secgroup_num),
'--server-groups',
str(compute_fakes.servgroup_num),
'--server-group-members',
str(compute_fakes.servgroup_members_num),
self.projects[0].name,
]
verifylist = [
('floating_ips', compute_fakes.floating_ip_num),
('fixed_ips', compute_fakes.fix_ip_num),
('injected_files', compute_fakes.injected_file_num),
('injected_file_content_bytes',
compute_fakes.injected_file_size_num),
(
'injected_file_content_bytes',
compute_fakes.injected_file_size_num,
),
('injected_file_path_bytes', compute_fakes.injected_path_size_num),
('key_pairs', compute_fakes.key_pair_num),
('cores', compute_fakes.core_num),
@ -709,8 +730,7 @@ class TestQuotaSet(TestQuota):
'floating_ips': compute_fakes.floating_ip_num,
'fixed_ips': compute_fakes.fix_ip_num,
'injected_files': compute_fakes.injected_file_num,
'injected_file_content_bytes':
compute_fakes.injected_file_size_num,
'injected_file_content_bytes': compute_fakes.injected_file_size_num, # noqa: E501
'injected_file_path_bytes': compute_fakes.injected_path_size_num,
'key_pairs': compute_fakes.key_pair_num,
'cores': compute_fakes.core_num,
@ -724,18 +744,22 @@ class TestQuotaSet(TestQuota):
}
self.compute_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs
self.projects[0].id, **kwargs
)
self.assertIsNone(result)
def test_quota_set_volume(self):
arglist = [
'--gigabytes', str(volume_fakes.QUOTA['gigabytes']),
'--snapshots', str(volume_fakes.QUOTA['snapshots']),
'--volumes', str(volume_fakes.QUOTA['volumes']),
'--backups', str(volume_fakes.QUOTA['backups']),
'--backup-gigabytes', str(volume_fakes.QUOTA['backup_gigabytes']),
'--gigabytes',
str(volume_fakes.QUOTA['gigabytes']),
'--snapshots',
str(volume_fakes.QUOTA['snapshots']),
'--volumes',
str(volume_fakes.QUOTA['volumes']),
'--backups',
str(volume_fakes.QUOTA['backups']),
'--backup-gigabytes',
str(volume_fakes.QUOTA['backup_gigabytes']),
'--per-volume-gigabytes',
str(volume_fakes.QUOTA['per_volume_gigabytes']),
self.projects[0].name,
@ -746,8 +770,10 @@ class TestQuotaSet(TestQuota):
('volumes', volume_fakes.QUOTA['volumes']),
('backups', volume_fakes.QUOTA['backups']),
('backup_gigabytes', volume_fakes.QUOTA['backup_gigabytes']),
('per_volume_gigabytes',
volume_fakes.QUOTA['per_volume_gigabytes']),
(
'per_volume_gigabytes',
volume_fakes.QUOTA['per_volume_gigabytes'],
),
('project', self.projects[0].name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -760,26 +786,31 @@ class TestQuotaSet(TestQuota):
'volumes': volume_fakes.QUOTA['volumes'],
'backups': volume_fakes.QUOTA['backups'],
'backup_gigabytes': volume_fakes.QUOTA['backup_gigabytes'],
'per_volume_gigabytes': volume_fakes.QUOTA['per_volume_gigabytes']
'per_volume_gigabytes': volume_fakes.QUOTA['per_volume_gigabytes'],
}
self.volume_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs
self.projects[0].id, **kwargs
)
self.assertIsNone(result)
def test_quota_set_volume_with_volume_type(self):
arglist = [
'--gigabytes', str(volume_fakes.QUOTA['gigabytes']),
'--snapshots', str(volume_fakes.QUOTA['snapshots']),
'--volumes', str(volume_fakes.QUOTA['volumes']),
'--backups', str(volume_fakes.QUOTA['backups']),
'--backup-gigabytes', str(volume_fakes.QUOTA['backup_gigabytes']),
'--gigabytes',
str(volume_fakes.QUOTA['gigabytes']),
'--snapshots',
str(volume_fakes.QUOTA['snapshots']),
'--volumes',
str(volume_fakes.QUOTA['volumes']),
'--backups',
str(volume_fakes.QUOTA['backups']),
'--backup-gigabytes',
str(volume_fakes.QUOTA['backup_gigabytes']),
'--per-volume-gigabytes',
str(volume_fakes.QUOTA['per_volume_gigabytes']),
'--volume-type', 'volume_type_backend',
'--volume-type',
'volume_type_backend',
self.projects[0].name,
]
verifylist = [
@ -788,8 +819,10 @@ class TestQuotaSet(TestQuota):
('volumes', volume_fakes.QUOTA['volumes']),
('backups', volume_fakes.QUOTA['backups']),
('backup_gigabytes', volume_fakes.QUOTA['backup_gigabytes']),
('per_volume_gigabytes',
volume_fakes.QUOTA['per_volume_gigabytes']),
(
'per_volume_gigabytes',
volume_fakes.QUOTA['per_volume_gigabytes'],
),
('volume_type', 'volume_type_backend'),
('project', self.projects[0].name),
]
@ -803,27 +836,34 @@ class TestQuotaSet(TestQuota):
'volumes_volume_type_backend': volume_fakes.QUOTA['volumes'],
'backups': volume_fakes.QUOTA['backups'],
'backup_gigabytes': volume_fakes.QUOTA['backup_gigabytes'],
'per_volume_gigabytes': volume_fakes.QUOTA['per_volume_gigabytes']
'per_volume_gigabytes': volume_fakes.QUOTA['per_volume_gigabytes'],
}
self.volume_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs
self.projects[0].id, **kwargs
)
self.assertIsNone(result)
def test_quota_set_network(self):
arglist = [
'--subnets', str(network_fakes.QUOTA['subnet']),
'--networks', str(network_fakes.QUOTA['network']),
'--floating-ips', str(network_fakes.QUOTA['floatingip']),
'--subnetpools', str(network_fakes.QUOTA['subnetpool']),
'--subnets',
str(network_fakes.QUOTA['subnet']),
'--networks',
str(network_fakes.QUOTA['network']),
'--floating-ips',
str(network_fakes.QUOTA['floatingip']),
'--subnetpools',
str(network_fakes.QUOTA['subnetpool']),
'--secgroup-rules',
str(network_fakes.QUOTA['security_group_rule']),
'--secgroups', str(network_fakes.QUOTA['security_group']),
'--routers', str(network_fakes.QUOTA['router']),
'--rbac-policies', str(network_fakes.QUOTA['rbac_policy']),
'--ports', str(network_fakes.QUOTA['port']),
'--secgroups',
str(network_fakes.QUOTA['security_group']),
'--routers',
str(network_fakes.QUOTA['router']),
'--rbac-policies',
str(network_fakes.QUOTA['rbac_policy']),
'--ports',
str(network_fakes.QUOTA['port']),
self.projects[0].name,
]
verifylist = [
@ -831,8 +871,10 @@ class TestQuotaSet(TestQuota):
('network', network_fakes.QUOTA['network']),
('floatingip', network_fakes.QUOTA['floatingip']),
('subnetpool', network_fakes.QUOTA['subnetpool']),
('security_group_rule',
network_fakes.QUOTA['security_group_rule']),
(
'security_group_rule',
network_fakes.QUOTA['security_group_rule'],
),
('security_group', network_fakes.QUOTA['security_group']),
('router', network_fakes.QUOTA['router']),
('rbac_policy', network_fakes.QUOTA['rbac_policy']),
@ -847,42 +889,56 @@ class TestQuotaSet(TestQuota):
'network': network_fakes.QUOTA['network'],
'floatingip': network_fakes.QUOTA['floatingip'],
'subnetpool': network_fakes.QUOTA['subnetpool'],
'security_group_rule':
network_fakes.QUOTA['security_group_rule'],
'security_group_rule': network_fakes.QUOTA['security_group_rule'],
'security_group': network_fakes.QUOTA['security_group'],
'router': network_fakes.QUOTA['router'],
'rbac_policy': network_fakes.QUOTA['rbac_policy'],
'port': network_fakes.QUOTA['port'],
}
self.network_mock.update_quota.assert_called_once_with(
self.projects[0].id,
**kwargs
self.projects[0].id, **kwargs
)
self.assertIsNone(result)
def test_quota_set_with_class(self):
arglist = [
'--injected-files', str(compute_fakes.injected_file_num),
'--injected-file-size', str(compute_fakes.injected_file_size_num),
'--injected-path-size', str(compute_fakes.injected_path_size_num),
'--key-pairs', str(compute_fakes.key_pair_num),
'--cores', str(compute_fakes.core_num),
'--ram', str(compute_fakes.ram_num),
'--instances', str(compute_fakes.instance_num),
'--properties', str(compute_fakes.property_num),
'--server-groups', str(compute_fakes.servgroup_num),
'--server-group-members', str(compute_fakes.servgroup_members_num),
'--gigabytes', str(compute_fakes.floating_ip_num),
'--snapshots', str(compute_fakes.fix_ip_num),
'--volumes', str(volume_fakes.QUOTA['volumes']),
'--network', str(network_fakes.QUOTA['network']),
'--injected-files',
str(compute_fakes.injected_file_num),
'--injected-file-size',
str(compute_fakes.injected_file_size_num),
'--injected-path-size',
str(compute_fakes.injected_path_size_num),
'--key-pairs',
str(compute_fakes.key_pair_num),
'--cores',
str(compute_fakes.core_num),
'--ram',
str(compute_fakes.ram_num),
'--instances',
str(compute_fakes.instance_num),
'--properties',
str(compute_fakes.property_num),
'--server-groups',
str(compute_fakes.servgroup_num),
'--server-group-members',
str(compute_fakes.servgroup_members_num),
'--gigabytes',
str(compute_fakes.floating_ip_num),
'--snapshots',
str(compute_fakes.fix_ip_num),
'--volumes',
str(volume_fakes.QUOTA['volumes']),
'--network',
str(network_fakes.QUOTA['network']),
'--class',
self.projects[0].name,
]
verifylist = [
('injected_files', compute_fakes.injected_file_num),
('injected_file_content_bytes',
compute_fakes.injected_file_size_num),
(
'injected_file_content_bytes',
compute_fakes.injected_file_size_num,
),
('injected_file_path_bytes', compute_fakes.injected_path_size_num),
('key_pairs', compute_fakes.key_pair_num),
('cores', compute_fakes.core_num),
@ -904,8 +960,7 @@ class TestQuotaSet(TestQuota):
kwargs_compute = {
'injected_files': compute_fakes.injected_file_num,
'injected_file_content_bytes':
compute_fakes.injected_file_size_num,
'injected_file_content_bytes': compute_fakes.injected_file_size_num, # noqa: E501
'injected_file_path_bytes': compute_fakes.injected_path_size_num,
'key_pairs': compute_fakes.key_pair_num,
'cores': compute_fakes.core_num,
@ -922,23 +977,26 @@ class TestQuotaSet(TestQuota):
}
self.compute_quotas_class_mock.update.assert_called_with(
self.projects[0].name,
**kwargs_compute
self.projects[0].name, **kwargs_compute
)
self.volume_quotas_class_mock.update.assert_called_with(
self.projects[0].name,
**kwargs_volume
self.projects[0].name, **kwargs_volume
)
self.assertNotCalled(self.network_mock.update_quota)
self.assertIsNone(result)
def test_quota_set_with_force(self):
arglist = [
'--cores', str(compute_fakes.core_num),
'--ram', str(compute_fakes.ram_num),
'--instances', str(compute_fakes.instance_num),
'--volumes', str(volume_fakes.QUOTA['volumes']),
'--subnets', str(network_fakes.QUOTA['subnet']),
'--cores',
str(compute_fakes.core_num),
'--ram',
str(compute_fakes.ram_num),
'--instances',
str(compute_fakes.instance_num),
'--volumes',
str(volume_fakes.QUOTA['volumes']),
'--subnets',
str(network_fakes.QUOTA['subnet']),
'--force',
self.projects[0].name,
]
@ -970,24 +1028,24 @@ class TestQuotaSet(TestQuota):
'force': True,
}
self.compute_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs_compute
self.projects[0].id, **kwargs_compute
)
self.volume_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs_volume
self.projects[0].id, **kwargs_volume
)
self.network_mock.update_quota.assert_called_once_with(
self.projects[0].id,
**kwargs_network
self.projects[0].id, **kwargs_network
)
self.assertIsNone(result)
def test_quota_set_with_no_force(self):
arglist = [
'--subnets', str(network_fakes.QUOTA['subnet']),
'--volumes', str(volume_fakes.QUOTA['volumes']),
'--cores', str(compute_fakes.core_num),
'--subnets',
str(network_fakes.QUOTA['subnet']),
'--volumes',
str(volume_fakes.QUOTA['volumes']),
'--cores',
str(compute_fakes.core_num),
'--no-force',
self.projects[0].name,
]
@ -1014,31 +1072,29 @@ class TestQuotaSet(TestQuota):
'check_limit': True,
}
self.compute_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs_compute
self.projects[0].id, **kwargs_compute
)
self.volume_quotas_mock.update.assert_called_once_with(
self.projects[0].id,
**kwargs_volume
self.projects[0].id, **kwargs_volume
)
self.network_mock.update_quota.assert_called_once_with(
self.projects[0].id,
**kwargs_network
self.projects[0].id, **kwargs_network
)
self.assertIsNone(result)
class TestQuotaShow(TestQuota):
def setUp(self):
super().setUp()
self.compute_quota = compute_fakes.FakeQuota.create_one_comp_quota()
self.compute_quotas_mock.get.return_value = self.compute_quota
self.compute_default_quota = \
self.compute_default_quota = (
compute_fakes.FakeQuota.create_one_default_comp_quota()
self.compute_quotas_mock.defaults.return_value = \
)
self.compute_quotas_mock.defaults.return_value = (
self.compute_default_quota
)
self.compute_quotas_class_mock.get.return_value = FakeQuotaResource(
None,
copy.deepcopy(compute_fakes.QUOTA),
@ -1048,8 +1104,9 @@ class TestQuotaShow(TestQuota):
self.volume_quota = volume_fakes.create_one_vol_quota()
self.volume_quotas_mock.get.return_value = self.volume_quota
self.volume_default_quota = volume_fakes.create_one_default_vol_quota()
self.volume_quotas_mock.defaults.return_value = \
self.volume_quotas_mock.defaults.return_value = (
self.volume_default_quota
)
self.volume_quotas_class_mock.get.return_value = FakeQuotaResource(
None,
copy.deepcopy(volume_fakes.QUOTA),
@ -1212,13 +1269,15 @@ class TestQuotaShow(TestQuota):
def test_quota_show__with_usage(self):
# update mocks to return detailed quota instead
self.compute_quota = \
self.compute_quota = (
compute_fakes.FakeQuota.create_one_comp_detailed_quota()
)
self.compute_quotas_mock.get.return_value = self.compute_quota
self.volume_quota = volume_fakes.create_one_detailed_quota()
self.volume_quotas_mock.get.return_value = self.volume_quota
self.network.get_quota.return_value = \
self.network.get_quota.return_value = (
network_fakes.FakeQuota.create_one_net_detailed_quota()
)
arglist = [
'--usage',