these patches are a rough PoC of a new machinery that abstracts option
handling/processing for installers ported on Jan's new installer
framework and also implement https://fedorahosted.org/freeipa/ticket/4517
The code is admittedly ugly but functional and demonstrated the base
architecture of the module. The user-facing entrypoints are the
FrontendManager function call and the individual Frontend
implementations which handle processing of options.
Here is a short example of the usage:
"""
# we want the Server installer to accept options from both CLI and
# config file
ServerCLIConfig = FrontendManager(Server, ConfigFrontend, CLIFrontend)
# add frontend-specific options to external option parser
# in case of CLIFrontend, all installer knobs are fed to the parser as #
options
# in case of ConfigFrontend, a single option is added which holds
# the location of config file
option_parser = IPAOptionParser()
ServerCLIConfig.add_options(option_parser)
# let the parser parse the command line
options, args = option_parser.parse_args(sys.argv[1:])
# get and validate all CLI/config options, instantiate the Server
# configurator and run installation
ServerCLIConfig.run(options, args)
"""
The second patch contains some basic tests for the new functionality
The third patch in the series plugs this functionality into the existing
CLI interface in a particularly nasty but (sorta-) functional way.
I would very much appreciate any of your comments concerning the whole
design and implementation. Suggestions to clean up the code are most
welcome.
--
Martin^3 Babinsky
From 881e3d58d492271a4d1c16e2833f992d7258fd28 Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Mon, 9 Nov 2015 18:17:52 +0100
Subject: [PATCH 3/3] POC: plug the new option handling machinery to
ipa-{server,replica}-install
This patch demonstrate the usage of the novel machinery for option processing
in the ipa-server-install and ipa-replica-install commands.
https://fedorahosted.org/freeipa/ticket/4517
---
install/tools/ipa-replica-install | 7 +-
install/tools/ipa-server-install | 4 +-
ipapython/install/cli.py | 212 ++++----------------------------------
3 files changed, 25 insertions(+), 198 deletions(-)
diff --git a/install/tools/ipa-replica-install b/install/tools/ipa-replica-install
index 60a853b419e503b5fb7fb1048034feb5c17afa02..c2f446aae011efbd12aaa593d2c9dd9a41d14d2a 100755
--- a/install/tools/ipa-replica-install
+++ b/install/tools/ipa-replica-install
@@ -18,15 +18,16 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
-from ipapython.install import cli
+from ipapython.install import cli, config
from ipaplatform.paths import paths
from ipaserver.install.server import Replica
ReplicaInstall = cli.install_tool(
- Replica,
+ config.FrontendManager(Replica, config.ConfigFileFrontend,
+ config.CLIFrontend,
+ positional_arguments=['replica_file']),
command_name='ipa-replica-install',
- positional_arguments='replica_file',
usage='%prog [options] REPLICA_FILE',
log_file_name=paths.IPAREPLICA_INSTALL_LOG,
debug_option=True,
diff --git a/install/tools/ipa-server-install b/install/tools/ipa-server-install
index 9fc78b538889fdfaca91d633e85398872c0fed0f..2cbd6b3c1fb3b807fb21f89ba2de670ca24e9973 100755
--- a/install/tools/ipa-server-install
+++ b/install/tools/ipa-server-install
@@ -21,12 +21,14 @@
#
from ipapython.install import cli
+from ipapython.install import config
from ipaplatform.paths import paths
from ipaserver.install.server import Server
ServerInstall = cli.install_tool(
- Server,
+ config.FrontendManager(Server, config.ConfigFileFrontend,
+ config.CLIFrontend),
command_name='ipa-server-install',
log_file_name=paths.IPASERVER_INSTALL_LOG,
debug_option=True,
diff --git a/ipapython/install/cli.py b/ipapython/install/cli.py
index d2250e51650b9de1c85473399e3462f42bf8770b..d778219ae0482509851727b44f2f0f049e7cafe1 100644
--- a/ipapython/install/cli.py
+++ b/ipapython/install/cli.py
@@ -15,7 +15,7 @@ import six
from ipapython import admintool, ipa_log_manager
from ipapython.ipautil import CheckedIPAddress, private_ccache
-from . import core, common
+from . import core, common, config
__all__ = ['install_tool', 'uninstall_tool']
@@ -23,8 +23,8 @@ if six.PY3:
long = int
-def install_tool(configurable_class, command_name, log_file_name,
- positional_arguments=None, usage=None, debug_option=False,
+def install_tool(frontend, command_name, log_file_name,
+ usage=None, debug_option=False,
use_private_ccache=True,
uninstall_log_file_name=None,
uninstall_positional_arguments=None, uninstall_usage=None):
@@ -32,10 +32,9 @@ def install_tool(configurable_class, command_name, log_file_name,
uninstall_positional_arguments is not None or
uninstall_usage is not None):
uninstall_kwargs = dict(
- configurable_class=configurable_class,
+ frontend=frontend,
command_name=command_name,
log_file_name=uninstall_log_file_name,
- positional_arguments=uninstall_positional_arguments,
usage=uninstall_usage,
debug_option=debug_option,
)
@@ -43,13 +42,12 @@ def install_tool(configurable_class, command_name, log_file_name,
uninstall_kwargs = None
return type(
- 'install_tool({0})'.format(configurable_class.__name__),
+ 'install_tool({0})'.format(frontend.configurable.__name__),
(InstallTool,),
dict(
- configurable_class=configurable_class,
+ frontend=frontend,
command_name=command_name,
log_file_name=log_file_name,
- positional_arguments=positional_arguments,
usage=usage,
debug_option=debug_option,
uninstall_kwargs=uninstall_kwargs,
@@ -58,13 +56,13 @@ def install_tool(configurable_class, command_name, log_file_name,
)
-def uninstall_tool(configurable_class, command_name, log_file_name,
+def uninstall_tool(frontend, command_name, log_file_name,
positional_arguments=None, usage=None, debug_option=False):
return type(
- 'uninstall_tool({0})'.format(configurable_class.__name__),
+ 'uninstall_tool({0})'.format(frontend.configurable.__name__),
(UninstallTool,),
dict(
- configurable_class=configurable_class,
+ frontend=frontend,
command_name=command_name,
log_file_name=log_file_name,
positional_arguments=positional_arguments,
@@ -75,7 +73,7 @@ def uninstall_tool(configurable_class, command_name, log_file_name,
class ConfigureTool(admintool.AdminTool):
- configurable_class = None
+ frontend = None
debug_option = False
positional_arguments = None
use_private_ccache = True
@@ -86,7 +84,7 @@ class ConfigureTool(admintool.AdminTool):
@classmethod
def add_options(cls, parser):
- transformed_cls = cls._transform(cls.configurable_class)
+ transformed_cls = cls._transform(cls.frontend.configurable)
if issubclass(transformed_cls, common.Interactive):
parser.add_option(
@@ -97,172 +95,14 @@ class ConfigureTool(admintool.AdminTool):
help="unattended (un)installation never prompts the user",
)
- basic_group = optparse.OptionGroup(parser, "basic options")
-
- groups = collections.OrderedDict()
- groups[None] = basic_group
-
- for owner_cls, name in transformed_cls.knobs():
- knob_cls = getattr(owner_cls, name)
- 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
- )
-
- for group, opt_group in groups.items():
- parser.add_option_group(opt_group)
+ cls.frontend.add_options(parser)
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
@@ -280,33 +120,17 @@ class ConfigureTool(admintool.AdminTool):
def run(self):
kwargs = {}
+ transformed_cls = self._transform(self.frontend.configurable)
+ self.frontend.configurable = transformed_cls
- transformed_cls = self._transform(self.configurable_class)
- knob_classes = {n: getattr(c, n) for c, n in transformed_cls.knobs()}
- for name in knob_classes:
- value = getattr(self.options, name, None)
- if value is not None:
- kwargs[name] = value
-
- if (issubclass(self.configurable_class, common.Interactive) and
+ if (issubclass(self.frontend.configurable, common.Interactive) and
not self.options.unattended):
kwargs['interactive'] = True
try:
- cfgr = transformed_cls(**kwargs)
- except core.KnobValueError as e:
- knob_cls = knob_classes[e.name]
- try:
- if self.positional_arguments is None:
- raise IndexError
- 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:
+ kwargs.update(self.frontend.get_kwargs(self.options, self.args))
+ cfgr = self.frontend.configurable(**kwargs)
+ except ValueError as e:
self.option_parser.error(str(e))
signal.signal(signal.SIGTERM, self.__signal_handler)
--
2.4.3
From 7c594bcbea0fbcc5f13c438799beb454bb34ff8e Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Tue, 27 Oct 2015 11:47:27 +0100
Subject: [PATCH 2/3] POC: test suite for the CLI/config option handlers
https://fedorahosted.org/freeipa/ticket/4517
---
ipatests/test_install/test_installtool_config.py | 434 +++++++++++++++++++++++
ipatests/util.py | 13 +
2 files changed, 447 insertions(+)
create mode 100644 ipatests/test_install/test_installtool_config.py
diff --git a/ipatests/test_install/test_installtool_config.py b/ipatests/test_install/test_installtool_config.py
new file mode 100644
index 0000000000000000000000000000000000000000..66e23794f849c0370c60142c3a88bcbff01eb6d8
--- /dev/null
+++ b/ipatests/test_install/test_installtool_config.py
@@ -0,0 +1,434 @@
+#
+# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
+#
+
+from copy import deepcopy
+import pytest
+from six.moves.configparser import RawConfigParser
+from csv import writer, QUOTE_NONE
+from cStringIO import StringIO
+import os
+import random
+import tempfile
+
+from ipapython.config import IPAOptionParser
+from ipapython.install import common, cli, config, core
+from ipapython.ipautil import CheckedIPAddress
+from ipatests.util import assert_error_msg
+
+
+def values_to_csv(values):
+# format the output as CSV string
+ outp_string = StringIO()
+ csv_writer = writer(outp_string, quoting=QUOTE_NONE,
+ escapechar='\\', lineterminator='')
+ csv_writer.writerow(list(str(v) for v in values))
+ result = outp_string.getvalue()
+ outp_string.close()
+ return result
+
+
+def args_from_kw(knobs, kwargs):
+ args = []
+ for owner_cls, name in knobs:
+ knob_cls = getattr(owner_cls, name)
+ option_name = '--{}'.format(name.replace('_', '-'))
+
+ if knob_cls.cli_short_name is not None:
+ option_name = '-{}'.format(knob_cls.cli_short_name)
+
+ try:
+ value = kwargs[name]
+ if isinstance(value, list):
+ for v in value:
+ args.extend([option_name, str(v)])
+ else:
+ args.extend([option_name, str(value)])
+ except KeyError:
+ pass
+
+ return args
+
+
+def config_from_kw(section, kwargs):
+ config_parser = RawConfigParser()
+ config_parser.add_section(section)
+
+ for kw in kwargs:
+ try:
+ value = kwargs[kw]
+ if isinstance(value, list):
+ config_parser.set(section, kw, values_to_csv(value))
+ else:
+ config_parser.set(section, kw, str(value))
+
+ except KeyError:
+ continue
+
+ filename = tempfile.mkstemp()[1]
+
+ with open(filename, 'w') as f:
+ config_parser.write(f)
+
+ return filename
+
+
+class MockInstaller(common.Installable, common.Interactive, core.Composite):
+
+ bool_knob = core.Knob(
+ bool,
+ False,
+ cli_short_name='b',
+ description='boolean option'
+ )
+
+ str_knob = core.Knob(
+ str,
+ 'string',
+ cli_short_name='s',
+ description='string option'
+ )
+
+ int_knob = core.Knob(
+ int,
+ 1,
+ cli_short_name='i',
+ description='integer option'
+ )
+
+ ip_knob = core.Knob(
+ 'ip',
+ None,
+ description='knob holding an IP address'
+ )
+
+# ip_local_knob = core.Knob(
+# 'ip-local',
+# None,
+# description='knob holding an IP address assigned to a local interface'
+# )
+
+ choice_knob = core.Knob(
+ {'choice1', 'choice2', 'choice3'},
+ 'choice1',
+ description='knob holding a choice between multiple values'
+ )
+
+ list_ints_knob = core.Knob(
+ (list, int),
+ None,
+ cli_short_name='l',
+ description='option storing list of integers'
+ )
+
+ list_strings_knob = core.Knob(
+ (list, str),
+ None,
+ cli_short_name='t',
+ description='option storing list of strings'
+ )
+
+ list_ips_knob = core.Knob(
+ (list, 'ip'),
+ None,
+ description='knob that holds multiple IP addresses'
+ )
+
+ validated_int_knob = core.Knob(
+ int,
+ 0,
+ cli_short_name='o',
+ description='option with custom validator'
+ )
+
+ @validated_int_knob.validator
+ def validated_int_knob(self, value):
+ if value < 0 or value > 10:
+ raise ValueError('invalid value {0}'.format(value))
+
+
+@pytest.fixture()
+def knobs(request):
+ return [(owner_cls, name) for owner_cls, name in MockInstaller.knobs()]
+
+
+@pytest.fixture()
+def frontend(request):
+ return config.FrontendManager(MockInstaller, config.ConfigFileFrontend,
+ config.CLIFrontend)
+
+
+@pytest.fixture()
+def parser(request):
+ return IPAOptionParser()
+
+
+@pytest.fixture()
+def installer(request):
+ return cli.install_tool(MockInstaller, 'mock-installer', None)
+
+
+@pytest.fixture()
+def valid_kw(request):
+ kwargs = {
+ 'bool_knob': True,
+ 'str_knob': 'string',
+ 'int_knob': 4,
+ 'ip_knob': CheckedIPAddress('1.2.3.4'),
+ 'choice_knob': 'choice2',
+ 'list_ints_knob': list(range(4)),
+ 'list_strings_knob': ['value1', 'value2', 'value3'],
+ 'list_ips_knob': [CheckedIPAddress('1.2.3.4'),
+ CheckedIPAddress('5.6.7.8'),
+ CheckedIPAddress('9.10.11.12')],
+ 'validated_int_knob': 4
+ }
+
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_bool(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['bool_knob'] = 'Trve'
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_int(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['int_knob'] = 'four'
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_ip(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['ip_knob'] = '371.342.231'
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_choice(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['choice_knob'] = 'choice4'
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_list(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['list_ints_knob'] = 'list'
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_int_in_list(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['list_ints_knob'] = [0, 'one', 2, 3]
+ return kwargs
+
+
+@pytest.fixture()
+def invalid_ip_in_list(request, valid_kw):
+ kwargs = deepcopy(valid_kw)
+ kwargs['list_ips_knob'] = ['1.2.3.4', '5.6.7', '9.10.11.12']
+ return kwargs
+
+
+@pytest.fixture()
+def valid_config(request, valid_kw):
+ filename = config_from_kw('global', valid_kw)
+
+ def fin():
+ try:
+ os.remove(filename)
+ except OSError:
+ pass
+
+ request.addfinalizer(fin)
+ return filename
+
+
+def assert_kwargs(expected, got):
+ for k in got:
+ assert k in expected
+ expected_val = expected[k]
+ if isinstance(expected_val, list):
+ for i, v in enumerate(expected_val):
+ assert v == got[k][i]
+ else:
+ assert expected_val == got[k]
+
+
+def get_kwargs(frontend, parser, arguments):
+ frontend.add_options(parser)
+ options, args = parser.parse_args(arguments)
+ return frontend.get_kwargs(options, args)
+
+
+class TestFrontend(object):
+ """
+ test basic frontend function
+ """
+ def test_valid_args(self, frontend, parser, knobs, valid_kw):
+ assert_kwargs(
+ valid_kw,
+ get_kwargs(
+ frontend,
+ parser,
+ args_from_kw(knobs, valid_kw)
+ )
+ )
+
+ def test_valid_config(self, frontend, parser, valid_config, valid_kw):
+ assert_kwargs(
+ valid_kw,
+ get_kwargs(
+ frontend,
+ parser,
+ ['--config', valid_config]
+ )
+ )
+
+ def test_mixed_cli_config(self, frontend, parser, knobs, valid_kw):
+ cli_kw = deepcopy(valid_kw)
+ config_kw = {}
+ num_config_kw = random.randint(0, len(cli_kw))
+
+ for i in range(num_config_kw):
+ k, v = cli_kw.popitem()
+ config_kw[k] = v
+
+ arguments = args_from_kw(knobs, cli_kw)
+ arguments.extend(
+ [
+ '--config',
+ config_from_kw('global', config_kw)
+ ]
+ )
+
+ assert_kwargs(
+ valid_kw,
+ get_kwargs(
+ frontend,
+ parser,
+ arguments
+ )
+ )
+
+ def test_cli_config_overriding(self, frontend, parser, knobs, valid_kw):
+ config_kw = dict(
+ int_knob=19,
+ list_ints_knob=list(range(15,20)),
+ list_ips_knob=['13.12.11.10']
+ )
+ valid_kw['int_knob'] = config_kw['int_knob']
+ arguments = args_from_kw(knobs, valid_kw)
+ arguments.extend(
+ [
+ '--config',
+ config_from_kw('global', config_kw)
+ ]
+ )
+
+ assert_kwargs(
+ valid_kw,
+ get_kwargs(
+ frontend,
+ parser,
+ arguments
+ )
+ )
+
+ def test_invalid_int(self, frontend, parser, knobs, invalid_int):
+ arguments = args_from_kw(knobs, invalid_int)
+
+ assert_error_msg(
+ ValueError,
+ "In CLI: option '{0}': invalid integer value '{1}'".format(
+ 'int_knob',
+ invalid_int['int_knob']
+ ),
+ get_kwargs,
+ frontend,
+ parser,
+ arguments
+ )
+
+ def test_invalid_ip(self, frontend, parser, knobs, invalid_ip):
+ arguments = args_from_kw(knobs, invalid_ip)
+
+ assert_error_msg(
+ ValueError,
+ "In CLI: option '{0}': invalid IP address '{1}'".format(
+ 'ip_knob',
+ invalid_ip['ip_knob']
+ ),
+ get_kwargs,
+ frontend,
+ parser,
+ arguments
+ )
+
+ def test_invalid_list(self, frontend, parser, knobs, invalid_list):
+ arguments = args_from_kw(knobs, invalid_list)
+
+ assert_error_msg(
+ ValueError,
+ "In CLI: option '{0}': invalid integer value '{1}'".format(
+ 'list_ints_knob',
+ invalid_list['list_ints_knob']
+ ),
+ get_kwargs,
+ frontend,
+ parser,
+ arguments
+ )
+
+ def test_invalid_choice(self, frontend, parser, knobs, invalid_choice):
+ arguments = args_from_kw(knobs, invalid_choice)
+
+ assert_error_msg(
+ ValueError,
+ "In CLI: option '{0}': invalid choice '{1}' "
+ "(choose from {2})".format(
+ 'choice_knob',
+ invalid_choice['choice_knob'],
+ ', '.join(repr(v) for v in MockInstaller.choice_knob.type)
+ ),
+ get_kwargs,
+ frontend,
+ parser,
+ arguments
+ )
+
+ def test_invalid_int_in_list(self, frontend, parser, knobs, invalid_int_in_list):
+ arguments = args_from_kw(knobs, invalid_int_in_list)
+
+ assert_error_msg(
+ ValueError,
+ "In CLI: option '{0}': invalid integer value '{1}'".format(
+ 'list_ints_knob',
+ invalid_int_in_list['list_ints_knob'][1]
+ ),
+ get_kwargs,
+ frontend,
+ parser,
+ arguments
+ )
+
+ def test_invalid_ip_in_list(self, frontend, parser, knobs, invalid_ip_in_list):
+ arguments = args_from_kw(knobs, invalid_ip_in_list)
+
+ assert_error_msg(
+ ValueError,
+ "In CLI: option '{0}': invalid IP address '{1}'".format(
+ 'list_ips_knob',
+ invalid_ip_in_list['list_ips_knob'][1]
+ ),
+ get_kwargs,
+ frontend,
+ parser,
+ arguments
+ )
diff --git a/ipatests/util.py b/ipatests/util.py
index c3c69816e264059102fbbb35bd91c8e7d51ded09..825a4a9b66cd8f6b8591e8bd74cd7379e10e77a1 100644
--- a/ipatests/util.py
+++ b/ipatests/util.py
@@ -381,6 +381,19 @@ def raises(exception, callback, *args, **kw):
return e
raise ExceptionNotRaised(exception)
+def assert_error_msg(exception, errmsg, callback, *args, **kwargs):
+ """
+ Tests whether an expected exception is raised and that the message contains
+ the expected substring
+
+ :param exception: exception class
+ :param errmsg: expected substring in error message
+ :param callback: callback function to be executed
+ :param args: function args
+ :param kwargs: function kwargs
+ """
+ e = raises(exception, callback, *args, **kwargs)
+ assert errmsg in str(e)
def getitem(obj, key):
"""
--
2.4.3
From 6b27d0644f62bed13c6715e047564784ff429824 Mon Sep 17 00:00:00 2001
From: Martin Babinsky <mbabi...@redhat.com>
Date: Wed, 21 Oct 2015 09:26:30 +0200
Subject: [PATCH 1/3] POC: modular processing of installer options from
multiple sources
This patch is a proof-of-concept preliminary implementation of an extensible
module which abstracts the passing of options to Configurable-based installers
from CLI and config files.
https://fedorahosted.org/freeipa/ticket/4517
---
ipapython/install/config.py | 480 ++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 480 insertions(+)
create mode 100644 ipapython/install/config.py
diff --git a/ipapython/install/config.py b/ipapython/install/config.py
new file mode 100644
index 0000000000000000000000000000000000000000..f9d4dbc35e322f536f9ba9b8a29f302a4b506a8c
--- /dev/null
+++ b/ipapython/install/config.py
@@ -0,0 +1,480 @@
+#
+# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
+#
+
+"""providers for installer configuration"""
+
+from __future__ import print_function
+import abc
+from copy import deepcopy
+from csv import reader, QUOTE_NONE
+
+import optparse
+import six
+from six.moves.configparser import (RawConfigParser, NoSectionError,
+ NoOptionError)
+
+from ipapython.install import core
+from ipapython.ipautil import CheckedIPAddress
+
+if six.PY3:
+ unicode = str
+ long = int
+
+
+_SCALAR_CONVERSION_MAP = {
+ str(int): {
+ 'conv': int,
+ 'name': 'integer value',
+ 'print_exc': False,
+ },
+ str(long): {
+ 'conv': int,
+ 'name': 'long integer value',
+ 'print_exc': False
+ },
+ str(str): {
+ 'conv': str,
+ 'name': 'string literal',
+ 'print_exc': False
+ },
+ str(bool): {
+ 'conv': bool,
+ 'name': 'Boolean value',
+ 'print_exc': False
+ },
+ 'ip': {
+ 'conv': CheckedIPAddress,
+ 'name': 'IP address',
+ 'print_exc': True
+ },
+ 'ip-local': {
+ 'conv': lambda x: CheckedIPAddress(x, match_local=True),
+ 'name': 'IP address',
+ 'print_exc': True,
+ },
+}
+
+
+def csv_from_string(string):
+ csv_reader = reader([string], skipinitialspace=True, strict=True,
+ quoting=QUOTE_NONE, escapechar='\\')
+ fields = csv_reader.next()
+ return list(field for field in fields)
+
+
+class _KnobValueConverterBase(object):
+ _list = False
+ _choice = {}
+ _scalar_conv_func = None
+ _scalar_name = ''
+ _print_conv_exc = False
+
+ @classmethod
+ def _handle_scalar_conversion_error(cls, value, exc):
+ err_msg = "invalid {0} {1}".format(cls._scalar_name, repr(value))
+
+ if cls._print_conv_exc:
+ err_msg = '{0}: {1}'.format(err_msg, exc)
+
+ raise ValueError(err_msg)
+
+ @classmethod
+ def _convert_scalar(cls, value):
+ try:
+ assert callable(cls._scalar_conv_func)
+ return cls._scalar_conv_func(value) # pylint: disable=E1102
+ except Exception as e:
+ cls._handle_scalar_conversion_error(value, e)
+
+ @classmethod
+ def _convert_list(cls, value):
+ if not isinstance(value, (list, tuple)):
+ raise ValueError("not a list/tuple")
+
+ return list(cls._convert_scalar(v) for v in value)
+
+ @classmethod
+ def _convert_choice(cls, value):
+ if value not in cls._choice:
+ raise ValueError(
+ "invalid choice {0} (choose from {1})".format(
+ repr(value), ', '.join(
+ sorted(repr(v) for v in cls._choice))))
+ return value
+
+ @classmethod
+ def convert(cls, value):
+ if cls._list:
+ return cls._convert_list(value)
+ elif cls._choice:
+ return cls._convert_choice(value)
+
+ return cls._convert_scalar(value)
+
+
+def _Converter(knob_type):
+ cls_dict = {}
+ cls_name = 'KnobValueConverter'
+ scalar_type = None
+ if isinstance(knob_type, tuple):
+ assert knob_type[0] is list
+ cls_dict['_list'] = True
+ scalar_type = str(knob_type[1])
+ elif isinstance(knob_type, set):
+ cls_dict['_choice'] = knob_type
+ else:
+ scalar_type = str(knob_type)
+
+ if scalar_type:
+ cls_dict['_scalar_conv_func'] = staticmethod(
+ _SCALAR_CONVERSION_MAP[scalar_type]['conv']
+ )
+ cls_dict['_scalar_name'] = _SCALAR_CONVERSION_MAP[scalar_type]['name']
+ cls_dict['_print_conv_exc'] = _SCALAR_CONVERSION_MAP[scalar_type]['print_exc']
+
+ return type(
+ cls_name,
+ (_KnobValueConverterBase,),
+ cls_dict
+ )
+
+
+class _FrontendManagerBase(object):
+ configurable = None
+ positional_arguments = None
+ _frontends = None
+ _converters = {}
+
+ @classmethod
+ def _bind_frontends(cls):
+ for f in cls._frontends:
+ f.frontend_manager = cls
+
+ @classmethod
+ def knob_dict(cls):
+ return {name: owner_cls for owner_cls, name
+ in cls.configurable.knobs()}
+
+ @classmethod
+ def _register_converters(cls, knob_dict):
+ # register all converters associated with configurable Knobs
+ for name, owner_cls in knob_dict.items():
+ knob_cls = getattr(owner_cls, name)
+ cls._converters[name] = _Converter(knob_cls.type)
+
+ @classmethod
+ def get_kwargs(cls, options, args):
+ processed_kwargs = {}
+ knobs = cls.knob_dict()
+ cls._register_converters(knobs)
+
+ for f in cls._frontends:
+ try:
+ frontend_kwargs = f.get_kwargs(options, args, knobs)
+ cls._convert_kwargs(frontend_kwargs)
+ except core.KnobValueError as e:
+ knob_cls = getattr(knobs[e.name], e.name)
+ option_name = f.knob_to_option(knob_cls, e.name)
+ raise ValueError(
+ "In {0}: option '{1}': {2}".format(
+ f.source_from, option_name, e
+ )
+ )
+ except Exception as e:
+ import traceback
+ traceback.print_exc(e)
+ raise RuntimeError(
+ 'In {0}: Unexpected error: {1}'.format(f.source_from, e)
+ )
+
+ processed_kwargs.update(frontend_kwargs)
+
+ return processed_kwargs
+
+ @classmethod
+ def _convert_kwargs(cls, kwargs):
+ for k, value in kwargs.items():
+ try:
+ kwargs[k] = cls._converters[k].convert(value)
+ except ValueError as e:
+ raise core.KnobValueError(
+ message=str(e),
+ name=k
+ )
+
+ @classmethod
+ def add_options(cls, option_parser):
+ knobs = cls.knob_dict()
+ for f in cls._frontends:
+ f.add_options(option_parser, knobs)
+
+ @classmethod
+ def run(cls, options, args):
+ kwargs = cls.get_kwargs(options, args)
+ cfgr = cls.configurable(**kwargs) # pylint: disable=E1102
+ cfgr.run()
+
+
+class FrontendBase(object):
+ """
+ Base class for all frontends
+ """
+
+ source_from = None
+ frontend_manager = None
+
+ @classmethod
+ @abc.abstractmethod
+ def add_options(cls, option_parser, knobs):
+ """
+ method for the provider to add options to the option parser instance
+ """
+ pass
+
+ @classmethod
+ @abc.abstractmethod
+ def get_kwargs(cls, options, args, knobs):
+ """
+ method to process parser options/arguments and return configurator
+ kwargs for further processing
+ """
+ return {}
+
+ @classmethod
+ def knob_to_option(cls, knob_cls, name):
+ """
+ use this method if you want to map a Knob to a custom option name.
+ """
+ return name
+
+
+class CLIFrontend(FrontendBase):
+ """
+ Provider of CLI options handled by OptionParser. Converts knobs to CLI
+ options and processes the parsed results
+ """
+
+ source_from = 'CLI'
+
+ @classmethod
+ def knob_to_option(cls, knob_cls, name):
+ cli_name = knob_cls.cli_name or name.replace('_', '-')
+ return '--{0}'.format(cli_name)
+
+ @classmethod
+ def add_options(cls, option_parser, knobs):
+ option_groups = {
+ None: optparse.OptionGroup(option_parser, "basic options")
+ }
+
+ positional_arguments = cls.frontend_manager.positional_arguments
+ for name, owner_cls in knobs.items():
+ if (positional_arguments is not None
+ and name in positional_arguments):
+ continue
+
+ knob_cls = getattr(owner_cls, name)
+
+ opt_group = cls._register_option_group(owner_cls, option_parser,
+ option_groups)
+
+ cls._add_option_to_group(knob_cls, name, opt_group)
+
+ for group_key, group_instance in option_groups.items():
+ option_parser.add_option_group(group_instance)
+
+ @staticmethod
+ def _register_option_group(owner_cls, option_parser, option_groups):
+ group_cls = owner_cls.group()
+ try:
+ opt_group = option_groups[group_cls]
+ except KeyError:
+ opt_group = option_groups[group_cls] = optparse.OptionGroup(
+ option_parser,
+ "{0} options".format(group_cls.description))
+
+ return opt_group
+
+ @classmethod
+ def _add_option_to_group(cls, knob_cls, name, opt_group):
+ kwargs = dict(
+ type='string',
+ dest=name,
+ action='store',
+ help=knob_cls.description
+ )
+
+ if knob_cls.type is bool:
+ kwargs.pop('type')
+ kwargs['action'] = 'store_true'
+ elif isinstance(knob_cls.type, tuple) and knob_cls.type[0] is list:
+ kwargs['action'] = 'append'
+
+ 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 = ''
+
+ opt_str = cls.knob_to_option(knob_cls, name)
+
+ if knob_cls.deprecated:
+ kwargs['help'] = optparse.SUPPRESS_HELP
+
+ opt_group.add_option(
+ short_opt_str, opt_str,
+ **kwargs
+ )
+
+ if knob_cls.cli_aliases:
+ kwargs['help'] = optparse.SUPPRESS_HELP
+ opt_strs = ['--{0}'.format(a) for a in knob_cls.cli_aliases]
+ opt_group.add_option(
+ *opt_strs,
+ **kwargs
+ )
+
+ @classmethod
+ def get_kwargs(cls, options, args, knobs):
+ kwargs = {}
+ for name in knobs:
+ value = getattr(options, name, None)
+ if value is not None:
+ kwargs[name] = value
+
+ if cls.frontend_manager.positional_arguments is not None:
+ kwargs.update(cls._get_positional_args(args))
+ return kwargs
+
+ @classmethod
+ def _get_positional_args(cls, args):
+ kwargs = {}
+ if len(args) > len(cls.frontend_manager.positional_arguments):
+ raise RuntimeError("Too many arguments provided")
+
+ for index, arg in enumerate(cls.frontend_manager.positional_arguments):
+ try:
+ kwargs[arg] = args[index]
+ except IndexError:
+ break
+
+ return kwargs
+
+
+class ConfigFileFrontend(FrontendBase):
+ """
+ Provider of Configurator kwargs from specified config file. Adds a CLI
+ option for config file
+ """
+ _section_name = 'global'
+ _config_file = None
+ _config_parser = None
+
+ source_from = "file '{}'"
+
+ @classmethod
+ def add_options(cls, option_parser, knobs):
+ option_parser.add_option(
+ "-c",
+ "--config",
+ dest='config',
+ default=None,
+ metavar='FILE',
+ help="get installer specific options from config file"
+ )
+
+ @classmethod
+ def get_kwargs(cls, options, args, knobs):
+
+ if options.config is None:
+ return {}
+
+ cls._config_file = options.config
+ cls.source_from = cls.source_from.format(cls._config_file)
+
+ cls._config_parser = RawConfigParser()
+ with open(cls._config_file) as f:
+ cls._config_parser.readfp(f)
+
+ cls._check_for_unknown_options(knobs)
+
+ kwargs = {}
+ for name in knobs:
+ knob_cls = getattr(knobs[name], name)
+
+ try:
+ kwargs[name] = cls._get_value(knob_cls, name)
+ except NoOptionError:
+ continue
+
+ return kwargs
+
+ @classmethod
+ def _check_for_unknown_options(cls, knobs):
+ config_options = {x for x in cls._config_parser.options(
+ cls._section_name)}
+ unknown_options = []
+
+ for c in config_options:
+ if c not in knobs:
+ unknown_options.append(c)
+
+ if unknown_options:
+ raise ValueError(
+ "unrecognized options: {}".format(
+ ', '.join(sorted(unknown_options))
+ )
+ )
+
+ @classmethod
+ def _get_value(cls, knob_cls, name):
+ if isinstance(knob_cls.type, tuple) and knob_cls.type[0] is list:
+ value = csv_from_string(
+ cls._config_parser.get(cls._section_name, name))
+ elif knob_cls.type is bool:
+ value = cls._config_parser.getboolean(cls._section_name, name)
+ else:
+ value = cls._config_parser.get(cls._section_name, name)
+
+ return value
+
+
+def FrontendManager(configurable, *frontends, **kwargs):
+ """
+ Binds selected frontends to the configurable class and returns a new
+ FrontendWrapper class implementing the option processing by specified
+ frontends
+ :param configurable: configurable class, e.g. Server
+ :param frontends: frontends which will be used by the resulting
+ FrontendWrapper class. The order of frontends determines the priority of
+ option processing (i.e. the first frontend is also the first to get
+ options)
+ :return: Frontend class inheriting from frontends and processing options
+ for the Configurable.
+ """
+ assert issubclass(configurable, core.Configurable)
+ for f in frontends:
+ assert issubclass(f, FrontendBase)
+
+ cls_name = 'FrontendManager({}, {})'.format(
+ configurable.__name__,
+ ', '.join(f.__name__ for f in frontends)
+ )
+ cls_dict = dict(
+ configurable=configurable,
+ _frontends=tuple(frontends)
+ )
+ cls_dict.update(kwargs)
+
+ cls = type(
+ cls_name,
+ (_FrontendManagerBase,),
+ cls_dict
+ )
+ cls._bind_frontends()
+ return cls
--
2.4.3
--
Manage your subscription for the Freeipa-devel mailing list:
https://www.redhat.com/mailman/listinfo/freeipa-devel
Contribute to FreeIPA: http://www.freeipa.org/page/Contribute/Code
