mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2026-10-24 22:27:41 -05:00
Petr Viktorin
3bf91eab25
Python 2 has keys()/values()/items(), which return lists, iterkeys()/itervalues()/iteritems(), which return iterators, and viewkeys()/viewvalues()/viewitems() which return views. Python 3 has only keys()/values()/items(), which return views. To get iterators, one can use iter() or a for loop/comprehension; for lists there's the list() constructor. When iterating through the entire dict, without modifying the dict, the difference between Python 2's items() and iteritems() is negligible, especially on small dicts (the main overhead is extra memory, not CPU time). In the interest of simpler code, this patch changes many instances of iteritems() to items(), iterkeys() to keys() etc. In other cases, helpers like six.itervalues are used. Reviewed-By: Christian Heimes <cheimes@redhat.com> Reviewed-By: Jan Cholasta <jcholast@redhat.com>
348 lines
12 KiB
Python
348 lines
12 KiB
Python
#
|
|
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
|
|
#
|
|
|
|
"""
|
|
Command line support.
|
|
"""
|
|
|
|
import collections
|
|
import optparse
|
|
import signal
|
|
|
|
from ipapython import admintool, ipa_log_manager
|
|
from ipapython.ipautil import CheckedIPAddress, private_ccache
|
|
|
|
from . import core, common
|
|
|
|
__all__ = ['install_tool', 'uninstall_tool']
|
|
|
|
|
|
def install_tool(configurable_class, command_name, log_file_name,
|
|
positional_arguments=None, usage=None, debug_option=False,
|
|
uninstall_log_file_name=None,
|
|
uninstall_positional_arguments=None, uninstall_usage=None):
|
|
if (uninstall_log_file_name is not None or
|
|
uninstall_positional_arguments is not None or
|
|
uninstall_usage is not None):
|
|
uninstall_kwargs = dict(
|
|
configurable_class=configurable_class,
|
|
command_name=command_name,
|
|
log_file_name=uninstall_log_file_name,
|
|
positional_arguments=uninstall_positional_arguments,
|
|
usage=uninstall_usage,
|
|
debug_option=debug_option,
|
|
)
|
|
else:
|
|
uninstall_kwargs = None
|
|
|
|
return type(
|
|
'install_tool({0})'.format(configurable_class.__name__),
|
|
(InstallTool,),
|
|
dict(
|
|
configurable_class=configurable_class,
|
|
command_name=command_name,
|
|
log_file_name=log_file_name,
|
|
positional_arguments=positional_arguments,
|
|
usage=usage,
|
|
debug_option=debug_option,
|
|
uninstall_kwargs=uninstall_kwargs,
|
|
)
|
|
)
|
|
|
|
|
|
def uninstall_tool(configurable_class, command_name, log_file_name,
|
|
positional_arguments=None, usage=None, debug_option=False):
|
|
return type(
|
|
'uninstall_tool({0})'.format(configurable_class.__name__),
|
|
(UninstallTool,),
|
|
dict(
|
|
configurable_class=configurable_class,
|
|
command_name=command_name,
|
|
log_file_name=log_file_name,
|
|
positional_arguments=positional_arguments,
|
|
usage=usage,
|
|
debug_option=debug_option,
|
|
)
|
|
)
|
|
|
|
|
|
class ConfigureTool(admintool.AdminTool):
|
|
configurable_class = None
|
|
debug_option = False
|
|
positional_arguments = None
|
|
|
|
@staticmethod
|
|
def _transform(configurable_class):
|
|
raise NotImplementedError
|
|
|
|
@classmethod
|
|
def add_options(cls, parser):
|
|
basic_group = optparse.OptionGroup(parser, "basic options")
|
|
|
|
groups = collections.OrderedDict()
|
|
groups[None] = basic_group
|
|
|
|
transformed_cls = cls._transform(cls.configurable_class)
|
|
for owner_cls, name in transformed_cls.knobs():
|
|
knob_cls = getattr(owner_cls, name)
|
|
if not knob_cls.initializable:
|
|
continue
|
|
if cls.positional_arguments and name in cls.positional_arguments:
|
|
continue
|
|
|
|
group_cls = owner_cls.group()
|
|
try:
|
|
opt_group = groups[group_cls]
|
|
except KeyError:
|
|
opt_group = groups[group_cls] = optparse.OptionGroup(
|
|
parser, "{0} options".format(group_cls.description))
|
|
|
|
kwargs = dict()
|
|
if knob_cls.type is bool:
|
|
kwargs['type'] = None
|
|
else:
|
|
kwargs['type'] = 'string'
|
|
kwargs['dest'] = name
|
|
kwargs['action'] = 'callback'
|
|
kwargs['callback'] = cls._option_callback
|
|
kwargs['callback_args'] = (knob_cls,)
|
|
if knob_cls.sensitive:
|
|
kwargs['sensitive'] = True
|
|
if knob_cls.cli_metavar:
|
|
kwargs['metavar'] = knob_cls.cli_metavar
|
|
|
|
if knob_cls.cli_short_name:
|
|
short_opt_str = '-{0}'.format(knob_cls.cli_short_name)
|
|
else:
|
|
short_opt_str = ''
|
|
cli_name = knob_cls.cli_name or name.replace('_', '-')
|
|
opt_str = '--{0}'.format(cli_name)
|
|
if not knob_cls.deprecated:
|
|
help = knob_cls.description
|
|
else:
|
|
help = optparse.SUPPRESS_HELP
|
|
opt_group.add_option(
|
|
short_opt_str, opt_str,
|
|
help=help,
|
|
**kwargs
|
|
)
|
|
|
|
if knob_cls.cli_aliases:
|
|
opt_strs = ['--{0}'.format(a) for a in knob_cls.cli_aliases]
|
|
opt_group.add_option(
|
|
*opt_strs,
|
|
help=optparse.SUPPRESS_HELP,
|
|
**kwargs
|
|
)
|
|
|
|
if issubclass(transformed_cls, common.Interactive):
|
|
basic_group.add_option(
|
|
'-U', '--unattended',
|
|
dest='unattended',
|
|
default=False,
|
|
action='store_true',
|
|
help="unattended (un)installation never prompts the user",
|
|
)
|
|
|
|
for group, opt_group in groups.items():
|
|
parser.add_option_group(opt_group)
|
|
|
|
super(ConfigureTool, cls).add_options(parser,
|
|
debug_option=cls.debug_option)
|
|
|
|
@classmethod
|
|
def _option_callback(cls, option, opt_str, value, parser, knob_cls):
|
|
old_value = getattr(parser.values, option.dest, None)
|
|
try:
|
|
value = cls._parse_knob(knob_cls, old_value, value)
|
|
except ValueError as e:
|
|
raise optparse.OptionValueError(
|
|
"option {0}: {1}".format(opt_str, e))
|
|
|
|
setattr(parser.values, option.dest, value)
|
|
|
|
@classmethod
|
|
def _parse_knob(cls, knob_cls, old_value, value):
|
|
if knob_cls.type is bool:
|
|
parse = bool
|
|
is_list = False
|
|
value = True
|
|
else:
|
|
if isinstance(knob_cls.type, tuple):
|
|
assert knob_cls.type[0] is list
|
|
value_type = knob_cls.type[1]
|
|
is_list = True
|
|
else:
|
|
value_type = knob_cls.type
|
|
is_list = False
|
|
|
|
if value_type is int:
|
|
def parse(value):
|
|
try:
|
|
return int(value, 0)
|
|
except ValueError:
|
|
raise ValueError(
|
|
"invalid integer value: {0}".format(repr(value)))
|
|
elif value_type is long:
|
|
def parse(value):
|
|
try:
|
|
return long(value, 0)
|
|
except ValueError:
|
|
raise ValueError(
|
|
"invalid long integer value: {0}".format(
|
|
repr(value)))
|
|
elif value_type == 'ip':
|
|
def parse(value):
|
|
try:
|
|
return CheckedIPAddress(value)
|
|
except Exception as e:
|
|
raise ValueError("invalid IP address {0}: {1}".format(
|
|
value, e))
|
|
elif value_type == 'ip-local':
|
|
def parse(value):
|
|
try:
|
|
return CheckedIPAddress(value, match_local=True)
|
|
except Exception as e:
|
|
raise ValueError("invalid IP address {0}: {1}".format(
|
|
value, e))
|
|
elif isinstance(value_type, set):
|
|
def parse(value):
|
|
if value not in value_type:
|
|
raise ValueError(
|
|
"invalid choice {0} (choose from {1})".format(
|
|
repr(value), ', '.join(repr(value_type))))
|
|
return value
|
|
else:
|
|
parse = value_type
|
|
|
|
value = parse(value)
|
|
|
|
if is_list:
|
|
old_value = old_value or []
|
|
old_value.append(value)
|
|
value = old_value
|
|
|
|
return value
|
|
|
|
def validate_options(self, needs_root=True):
|
|
super(ConfigureTool, self).validate_options(needs_root=needs_root)
|
|
|
|
if self.positional_arguments:
|
|
if len(self.args) > len(self.positional_arguments):
|
|
self.option_parser.error("Too many arguments provided")
|
|
|
|
index = 0
|
|
|
|
transformed_cls = self._transform(self.configurable_class)
|
|
for owner_cls, name in transformed_cls.knobs():
|
|
knob_cls = getattr(owner_cls, name)
|
|
if name not in self.positional_arguments:
|
|
continue
|
|
|
|
try:
|
|
value = self.args[index]
|
|
except IndexError:
|
|
break
|
|
|
|
old_value = getattr(self.options, name, None)
|
|
try:
|
|
value = self._parse_knob(knob_cls, old_value, value)
|
|
except ValueError as e:
|
|
self.option_parser.error(
|
|
"argument {0}: {1}".format(index + 1, e))
|
|
|
|
setattr(self.options, name, value)
|
|
|
|
index += 1
|
|
|
|
def _setup_logging(self, log_file_mode='w', no_file=False):
|
|
if no_file:
|
|
log_file_name = None
|
|
elif self.options.log_file:
|
|
log_file_name = self.options.log_file
|
|
else:
|
|
log_file_name = self.log_file_name
|
|
ipa_log_manager.standard_logging_setup(log_file_name,
|
|
debug=self.options.verbose)
|
|
self.log = ipa_log_manager.log_mgr.get_logger(self)
|
|
if log_file_name:
|
|
self.log.debug('Logging to %s' % log_file_name)
|
|
elif not no_file:
|
|
self.log.debug('Not logging to a file')
|
|
|
|
def run(self):
|
|
kwargs = {}
|
|
|
|
transformed_cls = self._transform(self.configurable_class)
|
|
for owner_cls, name in transformed_cls.knobs():
|
|
value = getattr(self.options, name, None)
|
|
if value is not None:
|
|
kwargs[name] = value
|
|
|
|
if (issubclass(self.configurable_class, common.Interactive) and
|
|
not self.options.unattended):
|
|
kwargs['interactive'] = True
|
|
|
|
try:
|
|
cfgr = transformed_cls(**kwargs)
|
|
except core.KnobValueError as e:
|
|
knob_cls = getattr(transformed_cls, e.name)
|
|
try:
|
|
index = self.positional_arguments.index(e.name)
|
|
except IndexError:
|
|
cli_name = knob_cls.cli_name or e.name.replace('_', '-')
|
|
desc = "option --{0}".format(cli_name)
|
|
else:
|
|
desc = "argument {0}".format(index + 1)
|
|
self.option_parser.error("{0}: {1}".format(desc, e))
|
|
except RuntimeError as e:
|
|
self.option_parser.error(str(e))
|
|
|
|
signal.signal(signal.SIGTERM, self.__signal_handler)
|
|
|
|
# Use private ccache
|
|
with private_ccache():
|
|
super(ConfigureTool, self).run()
|
|
|
|
cfgr.run()
|
|
|
|
@staticmethod
|
|
def __signal_handler(signum, frame):
|
|
raise KeyboardInterrupt
|
|
|
|
|
|
class InstallTool(ConfigureTool):
|
|
uninstall_kwargs = None
|
|
|
|
_transform = staticmethod(common.installer)
|
|
|
|
@classmethod
|
|
def add_options(cls, parser):
|
|
super(InstallTool, cls).add_options(parser)
|
|
|
|
if cls.uninstall_kwargs is not None:
|
|
uninstall_group = optparse.OptionGroup(parser, "uninstall options")
|
|
uninstall_group.add_option(
|
|
'--uninstall',
|
|
dest='uninstall',
|
|
default=False,
|
|
action='store_true',
|
|
help=("uninstall an existing installation. The uninstall can "
|
|
"be run with --unattended option"),
|
|
)
|
|
parser.add_option_group(uninstall_group)
|
|
|
|
@classmethod
|
|
def get_command_class(cls, options, args):
|
|
if cls.uninstall_kwargs is not None and options.uninstall:
|
|
uninstall_cls = uninstall_tool(**cls.uninstall_kwargs)
|
|
uninstall_cls.option_parser = cls.option_parser
|
|
return uninstall_cls
|
|
else:
|
|
return super(InstallTool, cls).get_command_class(options, args)
|
|
|
|
|
|
class UninstallTool(ConfigureTool):
|
|
_transform = staticmethod(common.uninstaller)
|