# Copyright (c) 2013 OpenStack Foundation # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import collections import os from urllib import parse import uuid import stevedore from cinderclient import exceptions def arg(*args, **kwargs): """Decorator for CLI args.""" def _decorator(func): add_arg(func, *args, **kwargs) return func return _decorator def exclusive_arg(group_name, *args, **kwargs): """Decorator for CLI mutually exclusive args.""" def _decorator(func): required = kwargs.pop('required', None) add_exclusive_arg(func, group_name, required, *args, **kwargs) return func return _decorator def env(*vars, **kwargs): """ returns the first environment variable set if none are non-empty, defaults to '' or keyword arg default """ for v in vars: value = os.environ.get(v, None) if value: return value return kwargs.get('default', '') def add_arg(f, *args, **kwargs): """Bind CLI arguments to a shell.py `do_foo` function.""" if not hasattr(f, 'arguments'): f.arguments = [] # NOTE(sirp): avoid dups that can occur when the module is shared across # tests. if (args, kwargs) not in f.arguments: # Because of the semantics of decorator composition if we just append # to the options list positional options will appear to be backwards. f.arguments.insert(0, (args, kwargs)) def add_exclusive_arg(f, group_name, required, *args, **kwargs): """Bind CLI mutally exclusive arguments to a shell.py `do_foo` function.""" if not hasattr(f, 'exclusive_args'): f.exclusive_args = collections.defaultdict(list) # Default required to False f.exclusive_args['__required__'] = collections.defaultdict(bool) # NOTE(sirp): avoid dups that can occur when the module is shared across # tests. if (args, kwargs) not in f.exclusive_args[group_name]: # Because of the semantics of decorator composition if we just append # to the options list positional options will appear to be backwards. f.exclusive_args[group_name].insert(0, (args, kwargs)) if required is not None: f.exclusive_args['__required__'][group_name] = required def unauthenticated(f): """ Adds 'unauthenticated' attribute to decorated function. Usage: @unauthenticated def mymethod(f): ... """ f.unauthenticated = True return f def isunauthenticated(f): """ Checks to see if the function is marked as not requiring authentication with the @unauthenticated decorator. Returns True if decorator is set to True, False otherwise. """ return getattr(f, 'unauthenticated', False) def build_query_param(params, sort=False): """parse list to url query parameters""" if not params: return "" if not sort: param_list = list(params.items()) else: param_list = list(sorted(params.items())) query_string = parse.urlencode( [(k, v) for (k, v) in param_list if v not in (None, '')]) # urllib's parse library used to adhere to RFC 2396 until # python 3.7. The library moved from RFC 2396 to RFC 3986 # for quoting URL strings in python 3.7 and '~' is now # included in the set of reserved characters. [1] # # Below ensures "~" is never encoded. See LP 1784728 [2] for more details. # [1] https://docs.python.org/3/library/urllib.parse.html#url-quoting # [2] https://bugs.launchpad.net/python-cinderclient/+bug/1784728 query_string = query_string.replace("%7E=", "~=") if query_string: query_string = "?%s" % (query_string,) return query_string def find_resource(manager, name_or_id, **kwargs): """Helper for the _find_* methods.""" is_group = kwargs.pop('is_group', False) # first try to get entity as integer id try: if isinstance(name_or_id, int) or name_or_id.isdigit(): if is_group: return manager.get(int(name_or_id), **kwargs) return manager.get(int(name_or_id)) except exceptions.NotFound: pass else: # now try to get entity as uuid try: uuid.UUID(name_or_id) if is_group: return manager.get(name_or_id, **kwargs) return manager.get(name_or_id) except (ValueError, exceptions.NotFound): pass try: try: resource = getattr(manager, 'resource_class', None) name_attr = resource.NAME_ATTR if resource else 'name' if is_group: kwargs[name_attr] = name_or_id return manager.find(**kwargs) return manager.find(**{name_attr: name_or_id}) except exceptions.NotFound: pass # finally try to find entity by human_id try: if is_group: kwargs['human_id'] = name_or_id return manager.find(**kwargs) return manager.find(human_id=name_or_id) except exceptions.NotFound: msg = "No %s with a name or ID of '%s' exists." % \ (manager.resource_class.__name__.lower(), name_or_id) raise exceptions.CommandError(msg) except exceptions.NoUniqueMatch: msg = ("Multiple %s matches found for '%s', use an ID to be more" " specific." % (manager.resource_class.__name__.lower(), name_or_id)) raise exceptions.CommandError(msg) def find_volume(cs, volume): """Get a volume by name or ID.""" return find_resource(cs.volumes, volume) def safe_issubclass(*args): """Like issubclass, but will just return False if not a class.""" try: if issubclass(*args): return True except TypeError: pass return False def _load_entry_point(ep_name, name=None): """Try to load the entry point ep_name that matches name.""" mgr = stevedore.NamedExtensionManager( namespace=ep_name, names=[name], # Ignore errors on load on_load_failure_callback=lambda mgr, entry_point, error: None, ) try: return mgr[name].plugin except KeyError: pass def get_function_name(func): return "%s.%s" % (func.__module__, func.__qualname__)