summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rwxr-xr-xipa5
-rw-r--r--ipa_server/mod_python_xmlrpc.py3
-rw-r--r--ipalib/cli.py351
-rw-r--r--ipalib/config.py221
-rw-r--r--ipalib/constants.py106
-rw-r--r--ipalib/frontend.py2
-rw-r--r--ipalib/load_plugins.py82
-rw-r--r--ipalib/plugable.py58
-rw-r--r--ipalib/plugins/b_xmlrpc.py23
-rw-r--r--ipalib/util.py86
-rwxr-xr-xlite-webui.py5
-rwxr-xr-xlite-xmlrpc.py80
-rwxr-xr-xmake-test1
-rw-r--r--tests/test_ipalib/test_cli.py208
-rw-r--r--tests/test_ipalib/test_config.py455
-rw-r--r--tests/test_ipalib/test_crud.py2
-rw-r--r--tests/test_ipalib/test_frontend.py10
-rw-r--r--tests/test_ipalib/test_plugable.py214
-rw-r--r--tests/util.py63
19 files changed, 1567 insertions, 408 deletions
diff --git a/ipa b/ipa
index b202a5d0..67e8d10c 100755
--- a/ipa
+++ b/ipa
@@ -28,8 +28,9 @@ The CLI functionality is implemented in ipalib/cli.py
import sys
from ipalib import api
from ipalib.cli import CLI
-import ipalib.load_plugins
if __name__ == '__main__':
- cli = CLI(api)
+ cli = CLI(api,
+ (s.decode('utf-8') for s in sys.argv[1:])
+ )
sys.exit(cli.run())
diff --git a/ipa_server/mod_python_xmlrpc.py b/ipa_server/mod_python_xmlrpc.py
index 814b1f65..450f4a51 100644
--- a/ipa_server/mod_python_xmlrpc.py
+++ b/ipa_server/mod_python_xmlrpc.py
@@ -46,11 +46,12 @@ from ipalib import config
from ipa_server import conn
from ipa_server.servercore import context
from ipa_server.servercore import ipautil
-import ipalib.load_plugins
from ipalib.util import xmlrpc_unmarshal
import string
+api.load_plugins()
+
# Global list of available functions
gfunctions = {}
diff --git a/ipalib/cli.py b/ipalib/cli.py
index a802f8ef..021e01ad 100644
--- a/ipalib/cli.py
+++ b/ipalib/cli.py
@@ -85,6 +85,8 @@ class help(frontend.Application):
self.application.build_parser(cmd).print_help()
+
+
class console(frontend.Application):
'Start the IPA interactive Python console.'
@@ -95,6 +97,21 @@ class console(frontend.Application):
)
+class env(frontend.Application):
+ """
+ Show environment variables.
+ """
+
+ def run(self):
+ return tuple(
+ (key, self.api.env[key]) for key in self.api.env
+ )
+
+ def output_for_cli(self, ret):
+ for (key, value) in ret:
+ print '%s = %r' % (key, value)
+
+
class show_api(text_ui):
'Show attributes on dynamic API object'
@@ -181,6 +198,7 @@ cli_application_commands = (
console,
show_api,
plugins,
+ env,
)
@@ -204,17 +222,160 @@ class KWCollector(object):
class CLI(object):
+ """
+ All logic for dispatching over command line interface.
+ """
+
__d = None
__mcl = None
- def __init__(self, api):
- self.__api = api
- self.__all_interactive = False
- self.__not_interactive = False
+ def __init__(self, api, argv):
+ self.api = api
+ self.argv = tuple(argv)
+ self.__done = set()
+
+ def run(self, init_only=False):
+ """
+ Parse ``argv`` and potentially run a command.
+
+ This method requires several initialization steps to be completed
+ first, all of which all automatically called with a single call to
+ `CLI.finalize()`. The initialization steps are broken into separate
+ methods simply to make it easy to write unit tests.
+
+ The initialization involves these steps:
+
+ 1. `CLI.parse_globals` parses the global options, which get stored
+ in ``CLI.options``, and stores the remaining args in
+ ``CLI.cmd_argv``.
+
+ 2. `CLI.bootstrap` initializes the environment information in
+ ``CLI.api.env``.
+
+ 3. `CLI.load_plugins` registers all plugins, including the
+ CLI-specific plugins.
+
+ 4. `CLI.finalize` instantiates all plugins and performs the
+ remaining initialization needed to use the `plugable.API`
+ instance.
+ """
+ self.__doing('run')
+ self.finalize()
+ if self.api.env.mode == 'unit_test':
+ return
+ if len(self.cmd_argv) < 1:
+ self.print_commands()
+ print 'Usage: ipa [global-options] COMMAND'
+ sys.exit(2)
+ key = self.cmd_argv[0]
+ if key not in self:
+ self.print_commands()
+ print 'ipa: ERROR: unknown command %r' % key
+ sys.exit(2)
+ return self.run_cmd(self[key])
+
+ def finalize(self):
+ """
+ Fully initialize ``CLI.api`` `plugable.API` instance.
+
+ This method first calls `CLI.load_plugins` to perform some dependant
+ initialization steps, after which `plugable.API.finalize` is called.
- def __get_api(self):
- return self.__api
- api = property(__get_api)
+ Finally, the CLI-specific commands are passed a reference to this
+ `CLI` instance by calling `frontend.Application.set_application`.
+ """
+ self.__doing('finalize')
+ self.load_plugins()
+ self.api.finalize()
+ for a in self.api.Application():
+ a.set_application(self)
+ assert self.__d is None
+ self.__d = dict(
+ (c.name.replace('_', '-'), c) for c in self.api.Command()
+ )
+
+ def load_plugins(self):
+ """
+ Load all standard plugins plus the CLI-specific plugins.
+
+ This method first calls `CLI.bootstrap` to preform some dependant
+ initialization steps, after which `plugable.API.load_plugins` is
+ called.
+
+ Finally, all the CLI-specific plugins are registered.
+ """
+ self.__doing('load_plugins')
+ self.bootstrap()
+ self.api.load_plugins()
+ for klass in cli_application_commands:
+ self.api.register(klass)
+
+ def bootstrap(self):
+ """
+ Initialize the ``CLI.api.env`` environment variables.
+
+ This method first calls `CLI.parse_globals` to perform some dependant
+ initialization steps. Then, using environment variables that may have
+ been passed in the global options, the ``overrides`` are constructed
+ and `plugable.API.bootstrap` is called.
+ """
+ self.__doing('bootstrap')
+ self.parse_globals()
+ self.api.env.verbose = self.options.verbose
+ if self.options.config_file:
+ self.api.env.conf = self.options.config_file
+ overrides = {}
+ if self.options.environment:
+ for a in self.options.environment.split(','):
+ a = a.split('=', 1)
+ if len(a) < 2:
+ parser.error('badly specified environment string,'\
+ 'use var1=val1[,var2=val2]..')
+ overrides[a[0].strip()] = a[1].strip()
+ overrides['context'] = 'cli'
+ self.api.bootstrap(**overrides)
+
+ def parse_globals(self):
+ """
+ Parse out the global options.
+
+ This method parses the global options out of the ``CLI.argv`` instance
+ attribute, after which two new instance attributes are available:
+
+ 1. ``CLI.options`` - an ``optparse.Values`` instance containing
+ the global options.
+
+ 2. ``CLI.cmd_argv`` - a tuple containing the remainder of
+ ``CLI.argv`` after the global options have been consumed.
+ """
+ self.__doing('parse_globals')
+ parser = optparse.OptionParser()
+ parser.disable_interspersed_args()
+ parser.add_option('-a', dest='prompt_all', action='store_true',
+ help='Prompt for all missing options interactively')
+ parser.add_option('-n', dest='interactive', action='store_false',
+ help='Don\'t prompt for any options interactively')
+ parser.add_option('-c', dest='config_file',
+ help='Specify different configuration file')
+ parser.add_option('-e', dest='environment',
+ help='Specify or override environment variables')
+ parser.add_option('-v', dest='verbose', action='store_true',
+ help='Verbose output')
+ parser.set_defaults(
+ prompt_all=False,
+ interactive=True,
+ verbose=False,
+ )
+ (options, args) = parser.parse_args(list(self.argv))
+ self.options = options
+ self.cmd_argv = tuple(args)
+
+ def __doing(self, name):
+ if name in self.__done:
+ raise StandardError(
+ '%s.%s() already called' % (self.__class__.__name__, name)
+ )
+ self.__done.add(name)
def print_commands(self):
std = set(self.api.Command) - set(self.api.Application)
@@ -234,66 +395,38 @@ class CLI(object):
cmd.doc,
)
- def __contains__(self, key):
- assert self.__d is not None, 'you must call finalize() first'
- return key in self.__d
-
- def __getitem__(self, key):
- assert self.__d is not None, 'you must call finalize() first'
- return self.__d[key]
-
- def finalize(self):
- api = self.api
- for klass in cli_application_commands:
- api.register(klass)
- api.finalize()
- for a in api.Application():
- a.set_application(self)
- self.build_map()
-
- def build_map(self):
- assert self.__d is None
- self.__d = dict(
- (c.name.replace('_', '-'), c) for c in self.api.Command()
- )
-
- def run(self):
- self.finalize()
- set_default_env(self.api.env)
- args = self.parse_globals()
- if len(args) < 1:
- self.print_commands()
- print 'Usage: ipa [global-options] COMMAND'
- sys.exit(2)
- key = args[0]
- if key not in self:
- self.print_commands()
- print 'ipa: ERROR: unknown command %r' % key
- sys.exit(2)
- return self.run_cmd(
- self[key],
- list(s.decode('utf-8') for s in args[1:])
- )
-
- def run_cmd(self, cmd, argv):
- kw = self.parse(cmd, argv)
+ def run_cmd(self, cmd):
+ kw = self.parse(cmd)
+ # If options.interactive, interactively validate params:
+ if self.options.interactive:
+ try:
+ kw = self.prompt_interactively(cmd, kw)
+ except KeyboardInterrupt:
+ return 0
+ # Now run the command
try:
- self.run_interactive(cmd, kw)
- except KeyboardInterrupt:
+ ret = cmd(**kw)
+ if callable(cmd.output_for_cli):
+ cmd.output_for_cli(ret)
return 0
- except errors.RuleError, e:
+ except StandardError, e:
print e
return 2
- return 0
- def run_interactive(self, cmd, kw):
+ def prompt_interactively(self, cmd, kw):
+ """
+ Interactively prompt for missing or invalid values.
+
+ By default this method will only prompt for *required* Param that
+ have a missing or invalid value. However, if
+ ``CLI.options.prompt_all`` is True, this method will prompt for any
+ params that have a missing or required values, even if the param is
+ optional.
+ """
for param in cmd.params():
if param.name not in kw:
- if not param.required:
- if not self.__all_interactive:
- continue
- elif self.__not_interactive:
- exit_error('Not enough arguments given')
+ if not (param.required or self.options.prompt_all):
+ continue
default = param.get_default(**kw)
if default is None:
prompt = '%s: ' % param.cli_name
@@ -311,29 +444,34 @@ class CLI(object):
break
except errors.ValidationError, e:
error = e.error
- if self.api.env.server_context:
- try:
- import krbV
- import ldap
- from ipa_server import conn
- from ipa_server.servercore import context
- krbccache = krbV.default_context().default_ccache().name
- context.conn = conn.IPAConn(self.api.env.ldaphost, self.api.env.ldapport, krbccache)
- except ImportError:
- print >> sys.stderr, "There was a problem importing a Python module: %s" % sys.exc_value
- return 2
- except ldap.LDAPError, e:
- print >> sys.stderr, "There was a problem connecting to the LDAP server: %s" % e[0].get('desc')
- return 2
- ret = cmd(**kw)
- if callable(cmd.output_for_cli):
- return cmd.output_for_cli(ret)
- else:
- return 0
+ return kw
- def parse(self, cmd, argv):
+# FIXME: This should be done as the plugins are loaded
+# if self.api.env.server_context:
+# try:
+# import krbV
+# import ldap
+# from ipa_server import conn
+# from ipa_server.servercore import context
+# krbccache = krbV.default_context().default_ccache().name
+# context.conn = conn.IPAConn(self.api.env.ldaphost, self.api.env.ldapport, krbccache)
+# except ImportError:
+# print >> sys.stderr, "There was a problem importing a Python module: %s" % sys.exc_value
+# return 2
+# except ldap.LDAPError, e:
+# print >> sys.stderr, "There was a problem connecting to the LDAP server: %s" % e[0].get('desc')
+# return 2
+# ret = cmd(**kw)
+# if callable(cmd.output_for_cli):
+# return cmd.output_for_cli(ret)
+# else:
+# return 0
+
+ def parse(self, cmd):
parser = self.build_parser(cmd)
- (kwc, args) = parser.parse_args(argv, KWCollector())
+ (kwc, args) = parser.parse_args(
+ list(self.cmd_argv[1:]), KWCollector()
+ )
kw = kwc.__todict__()
try:
arg_kw = cmd.args_to_kw(*args)
@@ -360,43 +498,6 @@ class CLI(object):
parser.add_option(o)
return parser
- def parse_globals(self, argv=sys.argv[1:]):
- parser = optparse.OptionParser()
- parser.disable_interspersed_args()
- parser.add_option('-a', dest='interactive', action='store_true',
- help='Prompt for all missing options interactively')
- parser.add_option('-n', dest='interactive', action='store_false',
- help='Don\'t prompt for any options interactively')
- parser.add_option('-c', dest='config_file',
- help='Specify different configuration file')
- parser.add_option('-e', dest='environment',
- help='Specify or override environment variables')
- parser.add_option('-v', dest='verbose', action='store_true',
- help='Verbose output')
- (options, args) = parser.parse_args(argv)
-
- if options.interactive == True:
- self.__all_interactive = True
- elif options.interactive == False:
- self.__not_interactive = True
- if options.verbose != None:
- self.api.env.verbose = True
- if options.environment:
- env_dict = {}
- for a in options.environment.split(','):
- a = a.split('=', 1)
- if len(a) < 2:
- parser.error('badly specified environment string,'\
- 'use var1=val1[,var2=val2]..')
- env_dict[a[0].strip()] = a[1].strip()
- self.api.env.update(env_dict, True)
- if options.config_file:
- self.api.env.update(read_config(options.config_file), True)
- else:
- self.api.env.update(read_config(), True)
-
- return args
-
def get_usage(self, cmd):
return ' '.join(self.get_usage_iter(cmd))
@@ -421,3 +522,17 @@ class CLI(object):
self.__mcl = max(len(k) for k in self.__d)
return self.__mcl
mcl = property(__get_mcl)
+
+ def isdone(self, name):
+ """
+ Return True in method named ``name`` has already been called.
+ """
+ return name in self.__done
+
+ def __contains__(self, key):
+ assert self.__d is not None, 'you must call finalize() first'
+ return key in self.__d
+
+ def __getitem__(self, key):
+ assert self.__d is not None, 'you must call finalize() first'
+ return self.__d[key]
diff --git a/ipalib/config.py b/ipalib/config.py
index 75e009dc..02a3fadd 100644
--- a/ipalib/config.py
+++ b/ipalib/config.py
@@ -25,11 +25,13 @@ It will also take care of settings that can be discovered by different
methods, such as DNS.
"""
-from ConfigParser import SafeConfigParser, ParsingError
+from ConfigParser import SafeConfigParser, ParsingError, RawConfigParser
import types
import os
-
+from os import path
+import sys
from errors import check_isinstance, raise_TypeError
+import constants
DEFAULT_CONF='/etc/ipa/ipa.conf'
@@ -126,6 +128,221 @@ class Environment(object):
return default
+class Env(object):
+ """
+ A mapping object used to store the environment variables.
+ """
+
+ __locked = False
+
+ def __init__(self):
+ object.__setattr__(self, '_Env__d', {})
+ object.__setattr__(self, '_Env__done', set())
+ self.ipalib = path.dirname(path.abspath(__file__))
+ self.site_packages = path.dirname(self.ipalib)
+ self.script = path.abspath(sys.argv[0])
+ self.bin = path.dirname(self.script)
+ self.home = path.abspath(os.environ['HOME'])
+ self.dot_ipa = path.join(self.home, '.ipa')
+
+ def __doing(self, name):
+ if name in self.__done:
+ raise StandardError(
+ '%s.%s() already called' % (self.__class__.__name__, name)
+ )
+ self.__done.add(name)
+
+ def __do_if_not_done(self, name):
+ if name not in self.__done:
+ getattr(self, name)()
+
+ def _isdone(self, name):
+ return name in self.__done
+
+ def _bootstrap(self, **overrides):
+ """
+ Initialize basic environment.
+
+ This method will initialize only enough environment information to
+ determine whether ipa is running in-tree, what the context is,
+ and the location of the configuration file.
+ """
+ self.__doing('_bootstrap')
+ for (key, value) in overrides.iteritems():
+ self[key] = value
+ if 'in_tree' not in self:
+ if self.bin == self.site_packages and \
+ path.isfile(path.join(self.bin, 'setup.py')):
+ self.in_tree = True
+ else:
+ self.in_tree = False
+ if 'context' not in self:
+ self.context = 'default'
+ if self.in_tree:
+ base = self.dot_ipa
+ else:
+ base = path.join('/', 'etc', 'ipa')
+ if 'conf' not in self:
+ self.conf = path.join(base, '%s.conf' % self.context)
+ if 'conf_default' not in self:
+ self.conf_default = path.join(base, 'default.conf')
+
+ def _finalize_core(self, **defaults):
+ """
+ Complete initialization of standard IPA environment.
+
+ After this method is called, the all environment variables
+ used by all the built-in plugins will be available.
+
+ This method should be called before loading any plugins. It will
+ automatically call `Env._bootstrap()` if it has not yet been called.
+
+ After this method has finished, the `Env` instance is still writable
+ so that third
+ """
+ self.__doing('_finalize_core')
+ self.__do_if_not_done('_bootstrap')
+ self._merge_config(self.conf)
+ if self.conf_default != self.conf:
+ self._merge_config(self.conf_default)
+ if 'in_server' not in self:
+ self.in_server = (self.context == 'server')
+ if 'log' not in self:
+ name = '%s.log' % self.context
+ if self.in_tree or self.context == 'cli':
+ self.log = path.join(self.dot_ipa, 'log', name)
+ else:
+ self.log = path.join('/', 'var', 'log', 'ipa', name)
+ for (key, value) in defaults.iteritems():
+ if key not in self:
+ self[key] = value
+
+ def _finalize(self, **lastchance):
+ """
+ Finalize and lock environment.
+
+ This method should be called after all plugins have bean loaded and
+ after `plugable.API.finalize()` has been called.
+ """
+ self.__doing('_finalize')
+ self.__do_if_not_done('_finalize_core')
+ for (key, value) in lastchance.iteritems():
+ if key not in self:
+ self[key] = value
+ self.__lock__()
+
+ def _merge_config(self, conf_file):
+ """
+ Merge values from ``conf_file`` into this `Env`.
+ """
+ section = constants.CONFIG_SECTION
+ if not path.isfile(conf_file):
+ return
+ parser = RawConfigParser()
+ try:
+ parser.read(conf_file)
+ except ParsingError:
+ return
+ if not parser.has_section(section):
+ parser.add_section(section)
+ items = parser.items(section)
+ if len(items) == 0:
+ return
+ i = 0
+ for (key, value) in items:
+ if key not in self:
+ self[key] = value
+ i += 1
+ return (i, len(items))
+
+ def __lock__(self):
+ """
+ Prevent further changes to environment.
+ """
+ if self.__locked is True:
+ raise StandardError(
+ '%s.__lock__() already called' % self.__class__.__name__
+ )
+ object.__setattr__(self, '_Env__locked', True)
+
+ def __islocked__(self):
+ return self.__locked
+
+ def __getattr__(self, name):
+ """
+ Return the attribute named ``name``.
+ """
+ if name in self.__d:
+ return self[name]
+ raise AttributeError('%s.%s' %
+ (self.__class__.__name__, name)
+ )
+
+ def __setattr__(self, name, value):
+ """
+ Set the attribute named ``name`` to ``value``.
+ """
+ self[name] = value
+
+ def __delattr__(self, name):
+ """
+ Raise AttributeError (deletion is not allowed).
+ """
+ raise AttributeError('cannot del %s.%s' %
+ (self.__class__.__name__, name)
+ )
+
+ def __getitem__(self, key):
+ """
+ Return the value corresponding to ``key``.
+ """
+ if key not in self.__d:
+ raise KeyError(key)
+ value = self.__d[key]
+ if callable(value):
+ return value()
+ return value
+
+ def __setitem__(self, key, value):
+ """
+ Set ``key`` to ``value``.
+ """
+ # FIXME: the key should be checked with check_name()
+ if self.__locked:
+ raise AttributeError('locked: cannot set %s.%s to %r' %
+ (self.__class__.__name__, key, value)
+ )
+ if key in self.__d or hasattr(self, key):
+ raise AttributeError('cannot overwrite %s.%s with %r' %
+ (self.__class__.__name__, key, value)
+ )
+ if not callable(value):
+ if isinstance(value, basestring):
+ value = str(value.strip())
+ if value.lower() == 'true':
+ value = True
+ elif value.lower() == 'false':
+ value = False
+ elif value.isdigit():
+ value = int(value)
+ assert type(value) in (str, int, bool)
+ object.__setattr__(self, key, value)
+ self.__d[key] = value
+
+ def __contains__(self, key):
+ """
+ Return True if instance contains ``key``; otherwise return False.
+ """
+ return key in self.__d
+
+ def __iter__(self): # Fix
+ """
+ Iterate through keys in ascending order.
+ """
+ for key in sorted(self.__d):
+ yield key
+
+
def set_default_env(env):
"""
Set default values for ``env``.
diff --git a/ipalib/constants.py b/ipalib/constants.py
new file mode 100644
index 00000000..f4a440c6
--- /dev/null
+++ b/ipalib/constants.py
@@ -0,0 +1,106 @@
+# Authors:
+# Martin Nagy <mnagy@redhat.com>
+# Jason Gerard DeRose <jderose@redhat.com>
+#
+# Copyright (C) 2008 Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; version 2 only
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+"""
+All constants centralized in one file.
+"""
+
+# The section to read in the config files, i.e. [global]
+CONFIG_SECTION = 'global'
+
+
+# The default configuration for api.env
+# This is a tuple instead of a dict so that it is immutable.
+# To create a dict with this config, just "d = dict(DEFAULT_CONFIG)".
+DEFAULT_CONFIG = (
+ # Domain, realm, basedn:
+ ('domain', 'example.com'),
+ ('realm', 'EXAMPLE.COM'),
+ ('basedn', 'dc=example,dc=com'),
+
+ # LDAP containers:
+ ('container_accounts', 'cn=accounts'),
+ ('container_user', 'cn=users,cn=accounts'),
+ ('container_group', 'cn=groups,cn=accounts'),
+ ('container_service', 'cn=services,cn=accounts'),
+ ('container_host', 'cn=computers,cn=accounts'),
+
+ # Ports, hosts, and URIs:
+ ('lite_xmlrpc_port', 8888),
+ ('lite_webui_port', 9999),
+ ('xmlrpc_uri', 'http://localhost:8888'),
+ ('ldap_uri', 'ldap://localhost:389'),
+ ('ldap_host', 'localhost'),
+ ('ldap_port', 389),
+
+ # Debugging:
+ ('verbose', False),
+ ('debug', False),
+ ('mode', 'production'),
+
+ # ********************************************************
+ # The remaining keys are never set from the values here!
+ # ********************************************************
+ #
+ # Env.__init__() or Env._bootstrap() or Env._finalize_core()
+ # will have filled in all the keys below by the time DEFAULT_CONFIG
+ # is merged in, so the values below are never actually used. They are
+ # listed both to provide a big picture and also so DEFAULT_CONFIG contains
+ # the keys that should be present after Env._finalize_core() is called.
+ #
+ # The values are all None so if for some reason any of these keys were
+ # set from the values here, an exception will be raised.
+
+ # Set in Env.__init__():
+ ('ipalib', None), # The directory containing ipalib/__init__.py
+ ('site_packages', None), # The directory contaning ipalib
+ ('script', None), # sys.argv[0]
+ ('bin', None), # The directory containing script
+ ('home', None), # The home directory of user underwhich process is running
+ ('dot_ipa', None), # ~/.ipa directory
+
+ # Set in Env._bootstrap():
+ ('in_tree', None), # Whether or not running in-tree (bool)
+ ('context', None), # Name of context, default is 'default'
+ ('conf', None), # Path to config file
+ ('conf_default', None), # Path to common default config file
+
+ # Set in Env._finalize_core():
+ ('in_server', None), # Whether or not running in-server (bool)
+ ('log', None), # Path to log file
+
+)
+
+
+LOGGING_CONSOLE_FORMAT = ': '.join([
+ '%(name)s',
+ '%(levelname)s',
+ '%(message)s',
+])
+
+
+# Tab-delimited format designed to be easily opened in a spreadsheet:
+LOGGING_FILE_FORMAT = ' '.join([
+ '%(created)f',
+ '%(levelname)s',
+ '%(message)r', # Using %r for repr() so message is a single line
+ '%(pathname)s',
+ '%(lineno)d',
+])
diff --git a/ipalib/frontend.py b/ipalib/frontend.py
index d918dd83..62a503cc 100644
--- a/ipalib/frontend.py
+++ b/ipalib/frontend.py
@@ -668,7 +668,7 @@ class Command(plugable.Plugin):
on the nearest IPA server and the actual work this command
performs is executed remotely.
"""
- if self.api.env.server_context:
+ if self.api.env.in_server:
target = self.execute
else:
target = self.forward
diff --git a/ipalib/load_plugins.py b/ipalib/load_plugins.py
deleted file mode 100644
index 4e02f5ba..00000000
--- a/ipalib/load_plugins.py
+++ /dev/null
@@ -1,82 +0,0 @@
-# Authors:
-# Jason Gerard DeRose <jderose@redhat.com>
-#
-# Copyright (C) 2008 Red Hat
-# see file 'COPYING' for use and warranty information
-#
-# This program is free software; you can redistribute it and/or
-# modify it under the terms of the GNU General Public License as
-# published by the Free Software Foundation; version 2 only
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-
-"""
-Importing this module causes the plugins to be loaded.
-
-This is not in __init__.py so that importing ipalib or its other sub-modules
-does not cause unnecessary side effects.
-
-Eventually this will also load the out-of tree plugins, but for now it just
-loads the internal plugins.
-"""
-
-import os
-from os import path
-import imp
-import inspect
-
-
-def find_modules_in_dir(src_dir):
- """
- Iterate through module names found in ``src_dir``.
- """
- if not (path.abspath(src_dir) == src_dir and path.isdir(src_dir)):
- return
- if path.islink(src_dir):
- return
- suffix = '.py'
- for name in sorted(os.listdir(src_dir)):
- if not name.endswith(suffix):
- continue
- py_file = path.join(src_dir, name)
- if path.islink(py_file) or not path.isfile(py_file):
- continue
- module = name[:-len(suffix)]
- if module == '__init__':
- continue
- yield module
-
-
-def load_plugins_in_dir(src_dir):
- """
- Import each Python module found in ``src_dir``.
- """
- for module in find_modules_in_dir(src_dir):
- imp.load_module(module, *imp.find_module(module, [src_dir]))
-
-
-def import_plugins(name):
- """
- Load all plugins found in standard 'plugins' sub-package.
- """
- try:
- plugins = __import__(name + '.plugins').plugins
- except ImportError:
- return
- src_dir = path.dirname(path.abspath(plugins.__file__))
- for name in find_modules_in_dir(src_dir):
- full_name = '%s.%s' % (plugins.__name__, name)
- __import__(full_name)
-
-
-for name in ['ipalib', 'ipa_server', 'ipa_not_a_package']:
- import_plugins(name)
-
-load_plugins_in_dir(path.expanduser('~/.freeipa'))
diff --git a/ipalib/plugable.py b/ipalib/plugable.py
index fd87586d..b0ba32b7 100644
--- a/ipalib/plugable.py
+++ b/ipalib/plugable.py
@@ -29,7 +29,9 @@ import re
import inspect
import errors
from errors import check_type, check_isinstance
-from config import Environment
+from config import Environment, Env
+import constants
+import util
class ReadOnly(object):
@@ -707,19 +709,67 @@ class API(DictProxy):
"""
Dynamic API object through which `Plugin` instances are accessed.
"""
- __finalized = False
def __init__(self, *allowed):
self.__d = dict()
+ self.__done = set()
self.register = Registrar(*allowed)
- self.env = Environment()
+ self.env = Env()
super(API, self).__init__(self.__d)
+ def __doing(self, name):
+ if name in self.__done:
+ raise StandardError(
+ '%s.%s() already called' % (self.__class__.__name__, name)
+ )
+ self.__done.add(name)
+
+ def __do_if_not_done(self, name):
+ if name not in self.__done:
+ getattr(self, name)()
+
+ def isdone(self, name):
+ return name in self.__done
+
+ def bootstrap(self, **overrides):
+ """
+ Initialize environment variables needed by built-in plugins.
+ """
+ self.__doing('bootstrap')
+ self.env._bootstrap(**overrides)
+ self.env._finalize_core(**dict(constants.DEFAULT_CONFIG))
+ if self.env.mode == 'unit_test':
+ return
+ logger = util.configure_logging(
+ self.env.log,
+ self.env.verbose,
+ )
+ object.__setattr__(self, 'logger', logger)
+
+ def load_plugins(self):
+ """
+ Load plugins from all standard locations.
+
+ `API.bootstrap` will automatically be called if it hasn't been
+ already.
+ """
+ self.__doing('load_plugins')
+ self.__do_if_not_done('bootstrap')
+ if self.env.mode == 'unit_test':
+ return
+ util.import_plugins_subpackage('ipalib')
+ if self.env.in_server:
+ util.import_plugins_subpackage('ipa_server')
+
def finalize(self):
"""
Finalize the registration, instantiate the plugins.
+
+ `API.bootstrap` will automatically be called if it hasn't been
+ already.
"""
- assert not self.__finalized, 'finalize() can only be called once'
+ self.__doing('finalize')
+ self.__do_if_not_done('bootstrap')
class PluginInstance(object):
"""
diff --git a/ipalib/plugins/b_xmlrpc.py b/ipalib/plugins/b_xmlrpc.py
index 572a7511..2c98fb8a 100644
--- a/ipalib/plugins/b_xmlrpc.py
+++ b/ipalib/plugins/b_xmlrpc.py
@@ -36,23 +36,26 @@ from ipalib import errors
class xmlrpc(Backend):
"""
- Kerberos backend plugin.
+ XML-RPC client backend plugin.
"""
- def get_client(self, verbose=False):
- # FIXME: The server uri should come from self.api.env.server_uri
- if api.env.get('kerberos'):
- server = api.env.server.next()
- if verbose: print "Connecting to %s" % server
- return xmlrpclib.ServerProxy('https://%s/ipa/xml' % server, transport=KerbTransport(), verbose=verbose)
- else:
- return xmlrpclib.ServerProxy('http://localhost:8888', verbose=verbose)
+ def get_client(self):
+ """
+ Return an xmlrpclib.ServerProxy instance (the client).
+ """
+ uri = self.api.env.xmlrpc_uri
+ if uri.startswith('https://'):
+ return xmlrpclib.ServerProxy(uri,
+ transport=KerbTransport(),
+ verbose=self.api.env.verbose,
+ )
+ return xmlrpclib.ServerProxy(uri, verbose=self.api.env.verbose)
def forward_call(self, name, *args, **kw):
"""
Forward a call over XML-RPC to an IPA server.
"""
- client = self.get_client(verbose=api.env.get('verbose', False))
+ client = self.get_client()
command = getattr(client, name)
params = xmlrpc_marshal(*args, **kw)
try:
diff --git a/ipalib/util.py b/ipalib/util.py
index 184c6d7c..d577524b 100644
--- a/ipalib/util.py
+++ b/ipalib/util.py
@@ -20,7 +20,14 @@
"""
Various utility functions.
"""
+
+import logging
+import os
+from os import path
+import imp
import krbV
+from constants import LOGGING_CONSOLE_FORMAT, LOGGING_FILE_FORMAT
+
def xmlrpc_marshal(*args, **kw):
"""
@@ -41,6 +48,7 @@ def xmlrpc_unmarshal(*params):
kw = {}
return (params[1:], kw)
+
def get_current_principal():
try:
return krbV.default_context().default_ccache().principal().name
@@ -48,3 +56,81 @@ def get_current_principal():
#TODO: do a kinit
print "Unable to get kerberos principal"
return None
+
+
+# FIXME: This function has no unit test
+def find_modules_in_dir(src_dir):
+ """
+ Iterate through module names found in ``src_dir``.
+ """
+ if not (path.abspath(src_dir) == src_dir and path.isdir(src_dir)):
+ return
+ if path.islink(src_dir):
+ return
+ suffix = '.py'
+ for name in sorted(os.listdir(src_dir)):
+ if not name.endswith(suffix):
+ continue
+ py_file = path.join(src_dir, name)
+ if path.islink(py_file) or not path.isfile(py_file):
+ continue
+ module = name[:-len(suffix)]
+ if module == '__init__':
+ continue
+ yield module
+
+
+# FIXME: This function has no unit test
+def load_plugins_in_dir(src_dir):
+ """
+ Import each Python module found in ``src_dir``.
+ """
+ for module in find_modules_in_dir(src_dir):
+ imp.load_module(module, *imp.find_module(module, [src_dir]))
+
+
+# FIXME: This function has no unit test
+def import_plugins_subpackage(name):
+ """
+ Import everythig in ``plugins`` sub-package of package named ``name``.
+ """
+ try:
+ plugins = __import__(name + '.plugins').plugins
+ except ImportError:
+ return
+ src_dir = path.dirname(path.abspath(plugins.__file__))
+ for name in find_modules_in_dir(src_dir):
+ full_name = '%s.%s' % (plugins.__name__, name)
+ __import__(full_name)
+
+
+def configure_logging(log_file, verbose):
+ """
+ Configure standard logging.
+ """
+ # Set logging level:
+ level = logging.INFO
+ if verbose:
+ level -= 10
+
+ log = logging.getLogger('ipa')
+ log.setLevel(level)
+
+ # Configure console handler
+ console = logging.StreamHandler()
+ console.setFormatter(logging.Formatter(LOGGING_CONSOLE_FORMAT))
+ log.addHandler(console)
+
+ # Configure file handler
+ log_dir = path.dirname(log_file)
+ if not path.isdir(log_dir):
+ try:
+ os.makedirs(log_dir)
+ except OSError:
+ log.warn('Could not create log_dir %r', log_dir)
+ return log
+ file_handler = logging.FileHandler(log_file)
+ file_handler.setFormatter(logging.Formatter(LOGGING_FILE_FORMAT))
+ log.addHandler(file_handler)
+
+ return log
diff --git a/lite-webui.py b/lite-webui.py
index ccef77ed..e75e5d9d 100755
--- a/lite-webui.py
+++ b/lite-webui.py
@@ -27,9 +27,8 @@ from cherrypy import expose, config, quickstart
from ipa_webui.templates import form, main
from ipa_webui import controller
from ipalib import api
-from ipalib import load_plugins
-
+api.load_plugins()
api.finalize()
@@ -42,5 +41,5 @@ class root(object):
setattr(self, cmd.name, ctr)
-if __name__ == '__main__'
+if __name__ == '__main__':
quickstart(root())
diff --git a/lite-xmlrpc.py b/lite-xmlrpc.py
index 7e9c69a8..3483ceb5 100755
--- a/lite-xmlrpc.py
+++ b/lite-xmlrpc.py
@@ -34,14 +34,10 @@ from ipalib import api
from ipalib import config
from ipa_server import conn
from ipa_server.servercore import context
-import ipalib.load_plugins
from ipalib.util import xmlrpc_unmarshal
import traceback
import krbV
-
-PORT=8888
-
class StoppableXMLRPCServer(SimpleXMLRPCServer.SimpleXMLRPCServer):
"""Override of TIME_WAIT"""
allow_reuse_address = True
@@ -65,31 +61,25 @@ class LoggingSimpleXMLRPCRequestHandler(SimpleXMLRPCServer.SimpleXMLRPCRequestHa
return (args, kw)
def _dispatch(self, method, params):
- """Dispatches the XML-RPC method.
+ """
+ Dispatches the XML-RPC method.
Methods beginning with an '_' are considered private and will
not be called.
"""
-
+ if method not in funcs:
+ logger.error('no such method %r', method)
+ raise Exception('method "%s" is not supported' % method)
+ func = funcs[method]
krbccache = krbV.default_context().default_ccache().name
-
- func = None
- try:
- try:
- # check to see if a matching function has been registered
- func = funcs[method]
- except KeyError:
- raise Exception('method "%s" is not supported' % method)
- (args, kw) = xmlrpc_unmarshal(*params)
- # FIXME: don't hardcode host and port
- context.conn = conn.IPAConn(api.env.ldaphost, api.env.ldapport, krbccache)
- logger.info("calling %s" % method)
- return func(*args, **kw)
- finally:
- # Clean up any per-request data and connections
-# for k in context.__dict__.keys():
-# del context.__dict__[k]
- pass
+ context.conn = conn.IPAConn(
+ api.env.ldap_host,
+ api.env.ldap_port,
+ krbccache,
+ )
+ logger.info('calling %s', method)
+ (args, kw) = xmlrpc_unmarshal(*params)
+ return func(*args, **kw)
def _marshaled_dispatch(self, data, dispatch_method = None):
try:
@@ -117,7 +107,7 @@ class LoggingSimpleXMLRPCRequestHandler(SimpleXMLRPCServer.SimpleXMLRPCRequestHa
def do_POST(self):
clientIP, port = self.client_address
- # Log client IP and Port
+ # Log client IP and Port
logger.info('Client IP: %s - Port: %s' % (clientIP, port))
try:
# get arguments
@@ -127,14 +117,14 @@ class LoggingSimpleXMLRPCRequestHandler(SimpleXMLRPCServer.SimpleXMLRPCRequestHa
params, method = xmlrpclib.loads(data)
# Log client request
- logger.info('Client request: \n%s\n' % data)
+ logger.info('Client request: \n%s\n' % data)
response = self._marshaled_dispatch(
data, getattr(self, '_dispatch', None))
- # Log server response
+ # Log server response
logger.info('Server response: \n%s\n' % response)
- except Exception, e:
+ except Exception, e:
# This should only happen if the module is buggy
# internal error, report as HTTP server error
print e
@@ -154,37 +144,29 @@ class LoggingSimpleXMLRPCRequestHandler(SimpleXMLRPCServer.SimpleXMLRPCRequestHa
if __name__ == '__main__':
- # Set up our logger
- logger = logging.getLogger('xmlrpcserver')
- hdlr = logging.FileHandler('xmlrpcserver.log')
- formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
- hdlr.setFormatter(formatter)
- logger.addHandler(hdlr)
- logger.setLevel(logging.INFO)
+ api.bootstrap(context='server')
+ api.load_plugins()
+ api.finalize()
+ logger = api.logger
# Set up the server
- XMLRPCServer = StoppableXMLRPCServer(("",PORT), LoggingSimpleXMLRPCRequestHandler)
-
+ XMLRPCServer = StoppableXMLRPCServer(
+ ('', api.env.lite_xmlrpc_port),
+ LoggingSimpleXMLRPCRequestHandler
+ )
XMLRPCServer.register_introspection_functions()
- api.finalize()
-
- # Initialize our environment
- config.set_default_env(api.env)
- env_dict = config.read_config()
- env_dict['server_context'] = True
- api.env.update(env_dict)
-
# Get and register all the methods
+
for cmd in api.Command:
- logger.info("registering %s" % cmd)
+ logger.debug('registering %s', cmd)
XMLRPCServer.register_function(api.Command[cmd], cmd)
-
funcs = XMLRPCServer.funcs
- print "Listening on port %d" % PORT
+ logger.info('Logging to file %r', api.env.log)
+ logger.info('Listening on port %d', api.env.lite_xmlrpc_port)
try:
XMLRPCServer.serve_forever()
except KeyboardInterrupt:
XMLRPCServer.server_close()
- print "Server shutdown."
+ logger.info('Server shutdown.')
diff --git a/make-test b/make-test
index 46456d6d..2d47707c 100755
--- a/make-test
+++ b/make-test
@@ -3,6 +3,7 @@
# Script to run nosetests under multiple versions of Python
versions="python2.4 python2.5 python2.6"
+versions="python2.5 python2.6"
for name in $versions
do
diff --git a/tests/test_ipalib/test_cli.py b/tests/test_ipalib/test_cli.py
index 50bfb932..7b3239d7 100644
--- a/tests/test_ipalib/test_cli.py
+++ b/tests/test_ipalib/test_cli.py
@@ -22,7 +22,8 @@ Test the `ipalib.cli` module.
"""
from tests.util import raises, getitem, no_set, no_del, read_only, ClassChecker
-from ipalib import cli, plugable
+from tests.util import TempHome
+from ipalib import cli, plugable, frontend, backend
def test_to_cli():
@@ -75,66 +76,185 @@ class DummyAPI(object):
pass
+config_cli = """
+[global]
+
+from_cli_conf = set in cli.conf
+"""
+
+config_default = """
+[global]
+
+from_default_conf = set in default.conf
+
+# Make sure cli.conf is loaded first:
+from_cli_conf = overridden in default.conf
+"""
+
+
+
+
class test_CLI(ClassChecker):
"""
Test the `ipalib.cli.CLI` class.
"""
_cls = cli.CLI
- def test_class(self):
+ def new(self, argv=tuple()):
+ home = TempHome()
+ api = plugable.API(
+ frontend.Command,
+ frontend.Object,
+ frontend.Method,
+ frontend.Property,
+ frontend.Application,
+ backend.Backend,
+ )
+ api.env.mode = 'unit_test'
+ api.env.in_tree = True
+ o = self.cls(api, argv)
+ assert o.api is api
+ return (o, api, home)
+
+ def check_cascade(self, *names):
+ (o, api, home) = self.new()
+ method = getattr(o, names[0])
+ for name in names:
+ assert o.isdone(name) is False
+ method()
+ for name in names:
+ assert o.isdone(name) is True
+ e = raises(StandardError, method)
+ assert str(e) == 'CLI.%s() already called' % names[0]
+
+ def test_init(self):
+ """
+ Test the `ipalib.cli.CLI.__init__` method.
+ """
+ argv = ['-v', 'user-add', '--first=Jonh', '--last=Doe']
+ (o, api, home) = self.new(argv)
+ assert o.api is api
+ assert o.argv == tuple(argv)
+
+ def test_run(self):
"""
- Test the `ipalib.cli.CLI` class.
+ Test the `ipalib.cli.CLI.run` method.
"""
- assert type(self.cls.api) is property
+ self.check_cascade(
+ 'run',
+ 'finalize',
+ 'load_plugins',
+ 'bootstrap',
+ 'parse_globals'
+ )
- def test_api(self):
+ def test_finalize(self):
"""
- Test the `ipalib.cli.CLI.api` property.
+ Test the `ipalib.cli.CLI.finalize` method.
"""
- api = 'the plugable.API instance'
- o = self.cls(api)
- assert read_only(o, 'api') is api
+ self.check_cascade(
+ 'finalize',
+ 'load_plugins',
+ 'bootstrap',
+ 'parse_globals'
+ )
- def dont_parse(self):
+ (o, api, home) = self.new()
+ assert api.isdone('finalize') is False
+ assert 'Command' not in api
+ o.finalize()
+ assert api.isdone('finalize') is True
+ assert list(api.Command) == \
+ sorted(k.__name__ for k in cli.cli_application_commands)
+
+ def test_load_plugins(self):
"""
- Test the `ipalib.cli.CLI.parse` method.
+ Test the `ipalib.cli.CLI.load_plugins` method.
"""
- o = self.cls(None)
- args = ['hello', 'naughty', 'nurse']
- kw = dict(
- first_name='Naughty',
- last_name='Nurse',
+ self.check_cascade(
+ 'load_plugins',
+ 'bootstrap',
+ 'parse_globals'
)
- opts = ['--%s=%s' % (k.replace('_', '-'), v) for (k, v) in kw.items()]
- assert o.parse(args + []) == (args, {})
- assert o.parse(opts + []) == ([], kw)
- assert o.parse(args + opts) == (args, kw)
- assert o.parse(opts + args) == (args, kw)
+ (o, api, home) = self.new()
+ assert api.isdone('load_plugins') is False
+ o.load_plugins()
+ assert api.isdone('load_plugins') is True
- def test_mcl(self):
+ def test_bootstrap(self):
"""
- Test the `ipalib.cli.CLI.mcl` property .
+ Test the `ipalib.cli.CLI.bootstrap` method.
"""
- cnt = 100
- api = DummyAPI(cnt)
- len(api.Command) == cnt
- o = self.cls(api)
- assert o.mcl is None
- o.build_map()
- assert o.mcl == 6 # len('cmd_99')
-
- def test_dict(self):
+ self.check_cascade(
+ 'bootstrap',
+ 'parse_globals'
+ )
+ # Test with empty argv
+ (o, api, home) = self.new()
+ keys = tuple(api.env)
+ assert api.isdone('bootstrap') is False
+ o.bootstrap()
+ assert api.isdone('bootstrap') is True
+ e = raises(StandardError, o.bootstrap)
+ assert str(e) == 'CLI.bootstrap() already called'
+ assert api.env.verbose is False
+ assert api.env.context == 'cli'
+ keys = tuple(api.env)
+ added = (
+ 'my_key',
+ 'whatever',
+ 'from_default_conf',
+ 'from_cli_conf'
+ )
+ for key in added:
+ assert key not in api.env
+ assert key not in keys
+
+ # Test with a populated argv
+ argv = ['-e', 'my_key=my_val,whatever=Hello']
+ (o, api, home) = self.new(argv)
+ home.write(config_default, '.ipa', 'default.conf')
+ home.write(config_cli, '.ipa', 'cli.conf')
+ o.bootstrap()
+ assert api.env.my_key == 'my_val'
+ assert api.env.whatever == 'Hello'
+ assert api.env.from_default_conf == 'set in default.conf'
+ assert api.env.from_cli_conf == 'set in cli.conf'
+ assert list(api.env) == sorted(keys + added)
+
+ def test_parse_globals(self):
"""
- Test container emulation of `ipalib.cli.CLI` class.
+ Test the `ipalib.cli.CLI.parse_globals` method.
"""
- cnt = 25
- api = DummyAPI(cnt)
- assert len(api.Command) == cnt
- o = self.cls(api)
- o.build_map()
- for cmd in api.Command():
- key = cli.to_cli(cmd.name)
- assert key in o
- assert o[key] is cmd
- assert cmd.name not in o
- raises(KeyError, getitem, o, cmd.name)
+ # Test with empty argv
+ (o, api, home) = self.new()
+ assert not hasattr(o, 'options')
+ assert not hasattr(o, 'cmd_argv')
+ assert o.isdone('parse_globals') is False
+ o.parse_globals()
+ assert o.isdone('parse_globals') is True
+ assert o.options.interactive is True
+ assert o.options.verbose is False
+ assert o.options.config_file is None
+ assert o.options.environment is None
+ assert o.cmd_argv == tuple()
+ e = raises(StandardError, o.parse_globals)
+ assert str(e) == 'CLI.parse_globals() already called'
+
+ # Test with a populated argv
+ argv = ('-a', '-n', '-v', '-c', '/my/config.conf', '-e', 'my_key=my_val')
+ cmd_argv = ('user-add', '--first', 'John', '--last', 'Doe')
+ (o, api, home) = self.new(argv + cmd_argv)
+ assert not hasattr(o, 'options')
+ assert not hasattr(o, 'cmd_argv')
+ assert o.isdone('parse_globals') is False
+ o.parse_globals()
+ assert o.isdone('parse_globals') is True
+ assert o.options.prompt_all is True
+ assert o.options.interactive is False
+ assert o.options.verbose is True
+ assert o.options.config_file == '/my/config.conf'
+ assert o.options.environment == 'my_key=my_val'
+ assert o.cmd_argv == cmd_argv
+ e = raises(StandardError, o.parse_globals)
+ assert str(e) == 'CLI.parse_globals() already called'
diff --git a/tests/test_ipalib/test_config.py b/tests/test_ipalib/test_config.py
index ed982a73..ddfbb708 100644
--- a/tests/test_ipalib/test_config.py
+++ b/tests/test_ipalib/test_config.py
@@ -22,10 +22,13 @@ Test the `ipalib.config` module.
"""
import types
-
-from tests.util import raises, setitem, delitem
-#from tests.util import getitem, setitem, delitem
-from ipalib import config
+import os
+from os import path
+import sys
+from tests.util import raises, setitem, delitem, ClassChecker
+from tests.util import getitem, setitem, delitem
+from tests.util import TempDir, TempHome
+from ipalib import config, constants
def test_Environment():
@@ -112,6 +115,450 @@ def test_Environment():
assert env.a != 1000
+# Random base64-encoded data to simulate a misbehaving config file.
+config_bad = """
+/9j/4AAQSkZJRgABAQEAlgCWAAD//gAIT2xpdmVy/9sAQwAQCwwODAoQDg0OEhEQExgoGhgWFhgx
+IyUdKDozPTw5Mzg3QEhcTkBEV0U3OFBtUVdfYmdoZz5NcXlwZHhcZWdj/8AACwgAlgB0AQERAP/E
+ABsAAAEFAQEAAAAAAAAAAAAAAAQAAQIDBQYH/8QAMhAAAgICAAUDBAIABAcAAAAAAQIAAwQRBRIh
+MUEGE1EiMmFxFIEVI0LBFjNSYnKRof/aAAgBAQAAPwDCtzmNRr1o/MEP1D6f7kdkRakgBsAtoQhk
+xls/y3Z113I11mhiUc1ewCf1Oq4anJgINdhLhQoextfedmYrenfcvdzaFQnYAE08XhONTWEK8+js
+Fpo1oqAKoAA8CWjoJJTHM8kJ5jsiOiszAKD1+IV/hmW76rosbfnlh1Pp3Mah2srCnXQE9YXiel/c
+p5r7uVj2CwxPTuFjjmdLbteNwmrLwsYe3TjsD8cmjKV43ycy+3o76D4llFuXmuCoZEPczXVOSsLv
+f5lgGpNZLxJL2jnvMar0/wAOp6jHDH/uO4RViY9f/KpRdfC6k3R9fRyj+pRZVkWKqF10e+hCKaFq
+XlH/ALlmhK7Met/uUGZ5ow8XL57lU8/Yt4lx4jUOJphLobTe/wDaHeZLxHXtJEya9o5lFzCqpmPY
+CUYoPtDfc9TLj0G5jZvHaMFirAs++oEHq9U4rbNiMp8a6wO/1Zbzn2alC+Nx8P1JfdeBboA+AILx
+rin8pfbA1ynvKuFUXZOXXkLbzOp2R56andL2G45MmO0RPWWLEe8GzaffoKb/ADI44Pt9ZXxAuuFa
+axtgp0BOSPCcviNX8n3Aw8KTNHB4FiY9StkobLWHVSeghq8M4bkAhKKyV6Hl8RV8MwMZG1Uuz3Jn
+IcUQJlMFGlJ6D4hfpymy7iChHKqvVtefxO7Ai1txLBIn7pcojN3jGVhQO0ZgCNfM5ZHycTLycSkr
+yhtqD4Bmrfw5cuqsm6xHXyp1seRLcHCp4dQy1bOzslj1MzeJ5dVFnuMVdgOiHxOWzrmyMg2Nrbde
+k3vR2OTddcd6A5R8GdZqOo67k4wXrLAQPMRKnzImMZEzm+P1nFz6cxQeVujagWR6jsYiqivlH/Ux
+1M+7jWY30i7QHx1gF11tjGyxiSfmVc+503pPidVROHYNNY21b/adVZZySo3uOo1qIZQYd9RCzfYm
+TUk/qW71LjGkTA+IYiZmM1T9N9j8Gee5+McXJem0/Wp8GUK6KOi7b5MgzFjsxpJHZGDKSCOxE3cD
+OvsxbbLc9lsT7Vc73KX4ln3q1ZyVrPx2J/uAjLyan37z7B+Zp4vqPJqKi0K4EvzvUt1qBMdfb+T5
+gycfzkXXuc35InfE6nO8Y9SjFc1Yqh2Hdj2mH/xFxR26XgD/AMRJf45mWMqW5bBD3KqAZlZtb++7
+kEqTsHe//sG1CcTBvy7OWpD+Sewhz8CyKCTYAQPiGV0LVWPdxqQNADQ6zL4nWq2gopU6+ofmA8x3
+1MlvfeIGbnBeCHitRt94IFbRGus2U9H08v13sT+BNHjeX/D4bY4OmP0rPPbHLMWJ2Yy2EDQjVsos
+BdeYDx8wo5L5KpSdLWPAE1+G8NrFtBKgOAXPTf6mzViql5ZBoE87eJZkKbOQ8m+Yjf5EBzcO621y
+GCqD0H41Obzq7U6vzM577HTXgzPPeOIvM1eB59nD8xXVj7bHTr8iej1MtlauvUMNgzi/V2ctliYy
+HYTq37nMExpZXRZYpZVJUdzNjg+FXYwZgdhv6nVVUJU/uH7iNf1CARrtF0IB113M7jTNVjFl2xJA
+5ROey88OrVOugOy67TDs+89NRKdSYILdRC8ZQVJ+PHyJs4fqe3EoFPLzBexPxOdusa2xndiWY7JM
+qMUNrzOTAfHC9XO9/E3vT9blVJB0o2Zu3MAoYrsL13Ii0Muw3XvJG9KkDOeqjf6gWcw5A33XN9nX
+tOeyMRFWy3Jch+bX7mXmCsW/5RBXUoHaOIRi2asAJ0IRbjqzll3o/EAaRiltDojgv2E1aePmhEWq
+rsNHZ7wir1K/8Y1vUCSCAd+IXiZ9b1gLYvN07trXTUD4rxN2TkUgEts8p2NDtD0t5MVGchr2Xe99
+hMPNvD1LX5J2TuZhGyYwBijjfiHU5bJXrnYfqBRtRtSbIBWG3+xI6HiLUWz8xA9RuaVNrMAPfB5x
+r6v9MLr4S1il7LaxyjY69Jl5eG+Kyhiv1jYIMGYMO8etGscKoJJ8Cbp4bVg4ivaq22t3G/tmRYo5
+zyjQ+JRFFET01GB0Yid9YiYh1l9KgEHqT8Tco/hewA/NzgdQdwTNGNTY3uU2crL9HN00ZlovNzfV
+oCanBrBRk1rpCHPUkQjjYoW4GtwAw30MDpuxvbAvpJceR5mXFFEY0W4o4mpg0XNXutQxPUHxLb8q
+7mRDyszLr6esz8u++9wL2LcvQb8RXCkhBV3A6mR5rEVSrdFPT8SBLMdsdmWe6P8AUAx+TB4oooxi
+i1Jmt0+5dfuOLbANB2H6MjzNzc2zv5ji1g2+5/MYnbb+Yh+T0kubUY940UUbUWtRpJN8w1CfebkK
+WfUu+/mDOAGOjsRo0UkIo+pPl6Rckl7ehuR1INGAj9u0kW2nXvK45YlQp1odukaICSAjgSQWf//Z
+"""
+
+# A config file that tries to override some standard vars:
+config_override = """
+[global]
+
+key0 = var0
+home = /home/sweet/home
+key1 = var1
+site_packages = planet
+key2 = var2
+key3 = var3
+"""
+
+# A config file that tests the automatic type conversion
+config_good = """
+[global]
+
+yes = tRUE
+no = fALse
+number = 42
+"""
+
+# A default config file to make sure it does not overwrite the explicit one
+config_default = """
+[global]
+
+yes = Hello
+not_in_other = foo_bar
+"""
+
+
+class test_Env(ClassChecker):
+ """
+ Test the `ipalib.config.Env` class.
+ """
+
+ _cls = config.Env
+
+ def new(self):
+ """
+ Set os.environ['HOME'] to a tempdir.
+
+ Returns tuple with new Env instance and the TempHome instance.
+ """
+ home = TempHome()
+ return (self.cls(), home)
+
+ def test_init(self):
+ """
+ Test the `ipalib.config.Env.__init__` method.
+ """
+ (o, home) = self.new()
+ ipalib = path.dirname(path.abspath(config.__file__))
+ assert o.ipalib == ipalib
+ assert o.site_packages == path.dirname(ipalib)
+ assert o.script == path.abspath(sys.argv[0])
+ assert o.bin == path.dirname(path.abspath(sys.argv[0]))
+ assert o.home == home.path
+ assert o.dot_ipa == home.join('.ipa')
+
+ def bootstrap(self, **overrides):
+ (o, home) = self.new()
+ assert o._isdone('_bootstrap') is False
+ o._bootstrap(**overrides)
+ assert o._isdone('_bootstrap') is True
+ e = raises(StandardError, o._bootstrap)
+ assert str(e) == 'Env._bootstrap() already called'
+ return (o, home)
+
+ def test_bootstrap(self):
+ """
+ Test the `ipalib.config.Env._bootstrap` method.
+ """
+ # Test defaults created by _bootstrap():
+ (o, home) = self.new()
+ assert 'in_tree' not in o
+ assert 'context' not in o
+ assert 'conf' not in o
+ o._bootstrap()
+ assert o.in_tree is False
+ assert o.context == 'default'
+ assert o.conf == '/etc/ipa/default.conf'
+ assert o.conf_default == o.conf
+
+ # Test overriding values created by _bootstrap()
+ (o, home) = self.bootstrap(in_tree='true', context='server')
+ assert o.in_tree is True
+ assert o.context == 'server'
+ assert o.conf == home.join('.ipa', 'server.conf')
+ (o, home) = self.bootstrap(conf='/my/wacky/whatever.conf')
+ assert o.in_tree is False
+ assert o.context == 'default'
+ assert o.conf == '/my/wacky/whatever.conf'
+ assert o.conf_default == '/etc/ipa/default.conf'
+ (o, home) = self.bootstrap(conf_default='/my/wacky/default.conf')
+ assert o.in_tree is False
+ assert o.context == 'default'
+ assert o.conf == '/etc/ipa/default.conf'
+ assert o.conf_default == '/my/wacky/default.conf'
+
+ # Test various overrides and types conversion
+ kw = dict(
+ yes=True,
+ no=False,
+ num=42,
+ msg='Hello, world!',
+ )
+ override = dict(
+ (k, u' %s ' % v) for (k, v) in kw.items()
+ )
+ (o, home) = self.new()
+ for key in kw:
+ assert key not in o
+ o._bootstrap(**override)
+ for (key, value) in kw.items():
+ assert getattr(o, key) == value
+ assert o[key] == value
+
+ def finalize_core(self, **defaults):
+ (o, home) = self.new()
+ assert o._isdone('_finalize_core') is False
+ o._finalize_core(**defaults)
+ assert o._isdone('_finalize_core') is True
+ e = raises(StandardError, o._finalize_core)
+ assert str(e) == 'Env._finalize_core() already called'
+ return (o, home)
+
+ def test_finalize_core(self):
+ """
+ Test the `ipalib.config.Env._finalize_core` method.
+ """
+ # Check that calls cascade up the chain:
+ (o, home) = self.new()
+ assert o._isdone('_bootstrap') is False
+ assert o._isdone('_finalize_core') is False
+ assert o._isdone('_finalize') is False
+ o._finalize_core()
+ assert o._isdone('_bootstrap') is True
+ assert o._isdone('_finalize_core') is True
+ assert o._isdone('_finalize') is False
+
+ # Check that it can't be called twice:
+ e = raises(StandardError, o._finalize_core)
+ assert str(e) == 'Env._finalize_core() already called'
+
+ # Check that _bootstrap() did its job:
+ (o, home) = self.bootstrap()
+ assert 'in_tree' in o
+ assert 'conf' in o
+ assert 'context' in o
+
+ # Check that keys _finalize_core() will set are not set yet:
+ assert 'log' not in o
+ assert 'in_server' not in o
+
+ # Check that _finalize_core() did its job:
+ o._finalize_core()
+ assert 'in_server' in o
+ assert 'log' in o
+ assert o.in_tree is False
+ assert o.context == 'default'
+ assert o.in_server is False
+ assert o.log == '/var/log/ipa/default.log'
+
+ # Check log is in ~/.ipa/log when context='cli'
+ (o, home) = self.bootstrap(context='cli')
+ o._finalize_core()
+ assert o.in_tree is False
+ assert o.log == home.join('.ipa', 'log', 'cli.log')
+
+ # Check **defaults can't set in_server nor log:
+ (o, home) = self.bootstrap(in_server='tRUE')
+ o._finalize_core(in_server=False)
+ assert o.in_server is True
+ (o, home) = self.bootstrap(log='/some/silly/log')
+ o._finalize_core(log='/a/different/log')
+ assert o.log == '/some/silly/log'
+
+ # Test loading config file, plus test some in-tree stuff
+ (o, home) = self.bootstrap(in_tree=True, context='server')
+ for key in ('yes', 'no', 'number'):
+ assert key not in o
+ home.write(config_good, '.ipa', 'server.conf')
+ home.write(config_default, '.ipa', 'default.conf')
+ o._finalize_core()
+ assert o.in_tree is True
+ assert o.context == 'server'
+ assert o.in_server is True
+ assert o.log == home.join('.ipa', 'log', 'server.log')
+ assert o.yes is True
+ assert o.no is False
+ assert o.number == 42
+ assert o.not_in_other == 'foo_bar'
+
+ # Test using DEFAULT_CONFIG:
+ defaults = dict(constants.DEFAULT_CONFIG)
+ (o, home) = self.finalize_core(**defaults)
+ assert list(o) == sorted(defaults)
+ for (key, value) in defaults.items():
+ if value is None:
+ continue
+ assert o[key] is value
+
+ def test_finalize(self):
+ """
+ Test the `ipalib.config.Env._finalize` method.
+ """
+ # Check that calls cascade up the chain:
+ (o, home) = self.new()
+ assert o._isdone('_bootstrap') is False
+ assert o._isdone('_finalize_core') is False
+ assert o._isdone('_finalize') is False
+ o._finalize()
+ assert o._isdone('_bootstrap') is True
+ assert o._isdone('_finalize_core') is True
+ assert o._isdone('_finalize') is True
+
+ # Check that it can't be called twice:
+ e = raises(StandardError, o._finalize)
+ assert str(e) == 'Env._finalize() already called'
+
+ # Check that _finalize() calls __lock__()
+ (o, home) = self.new()
+ assert o.__islocked__() is False
+ o._finalize()
+ assert o.__islocked__() is True
+ e = raises(StandardError, o.__lock__)
+ assert str(e) == 'Env.__lock__() already called'
+
+ # Check that **lastchance works
+ (o, home) = self.finalize_core()
+ key = 'just_one_more_key'
+ value = 'with one more value'
+ lastchance = {key: value}
+ assert key not in o
+ assert o._isdone('_finalize') is False
+ o._finalize(**lastchance)
+ assert key in o
+ assert o[key] is value
+
+ def test_merge_config(self):
+ """
+ Test the `ipalib.config.Env._merge_config` method.
+ """
+ tmp = TempDir()
+ assert callable(tmp.join)
+
+ # Test a config file that doesn't exist
+ no_exist = tmp.join('no_exist.conf')
+ assert not path.exists(no_exist)
+ o = self.cls()
+ keys = tuple(o)
+ orig = dict((k, o[k]) for k in o)
+ assert o._merge_config(no_exist) is None
+ assert tuple(o) == keys
+
+ # Test an empty config file
+ empty = tmp.touch('empty.conf')
+ assert path.isfile(empty)
+ assert o._merge_config(empty) is None
+ assert tuple(o) == keys
+
+ # Test a mal-formed config file:
+ bad = tmp.join('bad.conf')
+ open(bad, 'w').write(config_bad)
+ assert path.isfile(bad)
+ assert o._merge_config(bad) is None
+ assert tuple(o) == keys
+
+ # Test a valid config file that tries to override
+ override = tmp.join('override.conf')
+ open(override, 'w').write(config_override)
+ assert path.isfile(override)
+ assert o._merge_config(override) == (4, 6)
+ for (k, v) in orig.items():
+ assert o[k] is v
+ assert list(o) == sorted(keys + ('key0', 'key1', 'key2', 'key3'))
+ for i in xrange(4):
+ assert o['key%d' % i] == ('var%d' % i)
+ keys = tuple(o)
+
+ # Test a valid config file with type conversion
+ good = tmp.join('good.conf')
+ open(good, 'w').write(config_good)
+ assert path.isfile(good)
+ assert o._merge_config(good) == (3, 3)
+ assert list(o) == sorted(keys + ('yes', 'no', 'number'))
+ assert o.yes is True
+ assert o.no is False
+ assert o.number == 42
+
+ def test_lock(self):
+ """
+ Test the `ipalib.config.Env.__lock__` method.
+ """
+ o = self.cls()
+ assert o._Env__locked is False
+ o.__lock__()
+ assert o._Env__locked is True
+ e = raises(StandardError, o.__lock__)
+ assert str(e) == 'Env.__lock__() already called'
+
+ def test_getattr(self):
+ """
+ Test the `ipalib.config.Env.__getattr__` method.
+
+ Also tests the `ipalib.config.Env.__getitem__` method.
+ """
+ o = self.cls()
+ value = 'some value'
+ o.key = value
+ assert o.key is value
+ assert o['key'] is value
+ o.call = lambda: 'whatever'
+ assert o.call == 'whatever'
+ assert o['call'] == 'whatever'
+ for name in ('one', 'two'):
+ e = raises(AttributeError, getattr, o, name)
+ assert str(e) == 'Env.%s' % name
+ e = raises(KeyError, getitem, o, name)
+ assert str(e) == repr(name)
+
+ def test_setattr(self):
+ """
+ Test the `ipalib.config.Env.__setattr__` method.
+
+ Also tests the `ipalib.config.Env.__setitem__` method.
+ """
+ items = [
+ ('one', 1),
+ ('two', lambda: 2),
+ ('three', 3),
+ ('four', lambda: 4),
+ ]
+ for setvar in (setattr, setitem):
+ o = self.cls()
+ for (i, (name, value)) in enumerate(items):
+ setvar(o, name, value)
+ assert getattr(o, name) == i + 1
+ assert o[name] == i + 1
+ if callable(value):
+ assert name not in dir(o)
+ else:
+ assert name in dir(o)
+ e = raises(AttributeError, setvar, o, name, 42)
+ assert str(e) == 'cannot overwrite Env.%s with 42' % name
+ o = self.cls()
+ o.__lock__()
+ for (name, value) in items:
+ e = raises(AttributeError, setvar, o, name, value)
+ assert str(e) == \
+ 'locked: cannot set Env.%s to %r' % (name, value)
+ o = self.cls()
+ setvar(o, 'yes', ' true ')
+ assert o.yes is True
+ setvar(o, 'no', ' false ')
+ assert o.no is False
+ setvar(o, 'msg', u' Hello, world! ')
+ assert o.msg == 'Hello, world!'
+ assert type(o.msg) is str
+ setvar(o, 'num', ' 42 ')
+ assert o.num == 42
+
+ def test_delattr(self):
+ """
+ Test the `ipalib.config.Env.__delattr__` method.
+
+ This also tests that ``__delitem__`` is not implemented.
+ """
+ o = self.cls()
+ o.one = 1
+ assert o.one == 1
+ for key in ('one', 'two'):
+ e = raises(AttributeError, delattr, o, key)
+ assert str(e) == 'cannot del Env.%s' % key
+ e = raises(AttributeError, delitem, o, key)
+ assert str(e) == '__delitem__'
+
+ def test_contains(self):
+ """
+ Test the `ipalib.config.Env.__contains__` method.
+ """
+ o = self.cls()
+ items = [
+ ('one', 1),
+ ('two', lambda: 2),
+ ('three', 3),
+ ('four', lambda: 4),
+ ]
+ for (key, value) in items:
+ assert key not in o
+ o[key] = value
+ assert key in o
+
+ def test_iter(self):
+ """
+ Test the `ipalib.config.Env.__iter__` method.
+ """
+ o = self.cls()
+ default_keys = tuple(o)
+ keys = ('one', 'two', 'three', 'four', 'five')
+ for key in keys:
+ o[key] = 'the value'
+ assert list(o) == sorted(keys + default_keys)
+
+
def test_set_default_env():
"""
Test the `ipalib.config.set_default_env` function.
diff --git a/tests/test_ipalib/test_crud.py b/tests/test_ipalib/test_crud.py
index 794921aa..421eaca8 100644
--- a/tests/test_ipalib/test_crud.py
+++ b/tests/test_ipalib/test_crud.py
@@ -40,7 +40,7 @@ class CrudChecker(ClassChecker):
frontend.Method,
frontend.Property,
)
- config.set_default_env(api.env)
+ #config.set_default_env(api.env)
class user(frontend.Object):
takes_params = (
'givenname',
diff --git a/tests/test_ipalib/test_frontend.py b/tests/test_ipalib/test_frontend.py
index 966b5e93..5a678b5b 100644
--- a/tests/test_ipalib/test_frontend.py
+++ b/tests/test_ipalib/test_frontend.py
@@ -764,7 +764,8 @@ class test_Command(ClassChecker):
# Test in server context:
api = plugable.API(self.cls)
- api.env.update(dict(server_context=True))
+ #api.env.update(dict(server_context=True))
+ api.env.in_server = True
api.finalize()
o = my_cmd()
o.set_api(api)
@@ -774,7 +775,8 @@ class test_Command(ClassChecker):
# Test in non-server context
api = plugable.API(self.cls)
- api.env.update(dict(server_context=False))
+ #api.env.update(dict(server_context=False))
+ api.env.in_server = False
api.finalize()
o = my_cmd()
o.set_api(api)
@@ -907,7 +909,7 @@ class test_Object(ClassChecker):
frontend.Method,
frontend.Property,
)
- config.set_default_env(api.env)
+ #config.set_default_env(api.env)
api.finalize()
# Test with no primary keys:
@@ -964,7 +966,7 @@ class test_Object(ClassChecker):
frontend.Property,
backend.Backend,
)
- config.set_default_env(api.env)
+ #config.set_default_env(api.env)
class ldap(backend.Backend):
whatever = 'It worked!'
api.register(ldap)
diff --git a/tests/test_ipalib/test_plugable.py b/tests/test_ipalib/test_plugable.py
index 61011797..c7b8abbd 100644
--- a/tests/test_ipalib/test_plugable.py
+++ b/tests/test_ipalib/test_plugable.py
@@ -23,7 +23,7 @@ Test the `ipalib.plugable` module.
from tests.util import raises, no_set, no_del, read_only
from tests.util import getitem, setitem, delitem
-from tests.util import ClassChecker
+from tests.util import ClassChecker, TempHome
from ipalib import plugable, errors
@@ -764,99 +764,147 @@ def test_Registrar():
assert issubclass(klass, base)
-def test_API():
+class test_API(ClassChecker):
"""
Test the `ipalib.plugable.API` class.
"""
- assert issubclass(plugable.API, plugable.ReadOnly)
- # Setup the test bases, create the API:
- class base0(plugable.Plugin):
- __public__ = frozenset((
- 'method',
- ))
+ _cls = plugable.API
- def method(self, n):
- return n
+ def new(self, *bases):
+ home = TempHome()
+ api = self.cls(*bases)
+ api.env.mode = 'unit_test'
+ api.env.in_tree = True
+ return (api, home)
- class base1(plugable.Plugin):
- __public__ = frozenset((
- 'method',
- ))
+ def test_API(self):
+ """
+ Test the `ipalib.plugable.API` class.
+ """
+ assert issubclass(plugable.API, plugable.ReadOnly)
- def method(self, n):
- return n + 1
+ # Setup the test bases, create the API:
+ class base0(plugable.Plugin):
+ __public__ = frozenset((
+ 'method',
+ ))
- api = plugable.API(base0, base1)
- r = api.register
- assert isinstance(r, plugable.Registrar)
- assert read_only(api, 'register') is r
+ def method(self, n):
+ return n
- class base0_plugin0(base0):
- pass
- r(base0_plugin0)
+ class base1(plugable.Plugin):
+ __public__ = frozenset((
+ 'method',
+ ))
- class base0_plugin1(base0):
- pass
- r(base0_plugin1)
+ def method(self, n):
+ return n + 1
- class base0_plugin2(base0):
- pass
- r(base0_plugin2)
+ api = plugable.API(base0, base1)
+ r = api.register
+ assert isinstance(r, plugable.Registrar)
+ assert read_only(api, 'register') is r
- class base1_plugin0(base1):
- pass
- r(base1_plugin0)
+ class base0_plugin0(base0):
+ pass
+ r(base0_plugin0)
- class base1_plugin1(base1):
- pass
- r(base1_plugin1)
+ class base0_plugin1(base0):
+ pass
+ r(base0_plugin1)
- class base1_plugin2(base1):
- pass
- r(base1_plugin2)
-
- # Test API instance:
- api.finalize()
-
- def get_base(b):
- return 'base%d' % b
-
- def get_plugin(b, p):
- return 'base%d_plugin%d' % (b, p)
-
- for b in xrange(2):
- base_name = get_base(b)
- ns = getattr(api, base_name)
- assert isinstance(ns, plugable.NameSpace)
- assert read_only(api, base_name) is ns
- assert len(ns) == 3
- for p in xrange(3):
- plugin_name = get_plugin(b, p)
- proxy = ns[plugin_name]
- assert isinstance(proxy, plugable.PluginProxy)
- assert proxy.name == plugin_name
- assert read_only(ns, plugin_name) is proxy
- assert read_only(proxy, 'method')(7) == 7 + b
-
- # Test that calling finilize again raises AssertionError:
- raises(AssertionError, api.finalize)
-
- # Test with base class that doesn't request a proxy
- class NoProxy(plugable.Plugin):
- __proxy__ = False
- api = plugable.API(NoProxy)
- class plugin0(NoProxy):
- pass
- api.register(plugin0)
- class plugin1(NoProxy):
- pass
- api.register(plugin1)
- api.finalize()
- names = ['plugin0', 'plugin1']
- assert list(api.NoProxy) == names
- for name in names:
- plugin = api.NoProxy[name]
- assert getattr(api.NoProxy, name) is plugin
- assert isinstance(plugin, plugable.Plugin)
- assert not isinstance(plugin, plugable.PluginProxy)
+ class base0_plugin2(base0):
+ pass
+ r(base0_plugin2)
+
+ class base1_plugin0(base1):
+ pass
+ r(base1_plugin0)
+
+ class base1_plugin1(base1):
+ pass
+ r(base1_plugin1)
+
+ class base1_plugin2(base1):
+ pass
+ r(base1_plugin2)
+
+ # Test API instance:
+ assert api.isdone('bootstrap') is False
+ assert api.isdone('finalize') is False
+ api.finalize()
+ assert api.isdone('bootstrap') is True
+ assert api.isdone('finalize') is True
+
+ def get_base(b):
+ return 'base%d' % b
+
+ def get_plugin(b, p):
+ return 'base%d_plugin%d' % (b, p)
+
+ for b in xrange(2):
+ base_name = get_base(b)
+ ns = getattr(api, base_name)
+ assert isinstance(ns, plugable.NameSpace)
+ assert read_only(api, base_name) is ns
+ assert len(ns) == 3
+ for p in xrange(3):
+ plugin_name = get_plugin(b, p)
+ proxy = ns[plugin_name]
+ assert isinstance(proxy, plugable.PluginProxy)
+ assert proxy.name == plugin_name
+ assert read_only(ns, plugin_name) is proxy
+ assert read_only(proxy, 'method')(7) == 7 + b
+
+ # Test that calling finilize again raises AssertionError:
+ e = raises(StandardError, api.finalize)
+ assert str(e) == 'API.finalize() already called', str(e)
+
+ # Test with base class that doesn't request a proxy
+ class NoProxy(plugable.Plugin):
+ __proxy__ = False
+ api = plugable.API(NoProxy)
+ class plugin0(NoProxy):
+ pass
+ api.register(plugin0)
+ class plugin1(NoProxy):
+ pass
+ api.register(plugin1)
+ api.finalize()
+ names = ['plugin0', 'plugin1']
+ assert list(api.NoProxy) == names
+ for name in names:
+ plugin = api.NoProxy[name]
+ assert getattr(api.NoProxy, name) is plugin
+ assert isinstance(plugin, plugable.Plugin)
+ assert not isinstance(plugin, plugable.PluginProxy)
+
+ def test_bootstrap(self):
+ """
+ Test the `ipalib.plugable.API.bootstrap` method.
+ """
+ (o, home) = self.new()
+ assert o.env._isdone('_bootstrap') is False
+ assert o.env._isdone('_finalize_core') is False
+ assert o.isdone('bootstrap') is False
+ o.bootstrap(my_test_override='Hello, world!')
+ assert o.isdone('bootstrap') is True
+ assert o.env._isdone('_bootstrap') is True
+ assert o.env._isdone('_finalize_core') is True
+ assert o.env.my_test_override == 'Hello, world!'
+ e = raises(StandardError, o.bootstrap)
+ assert str(e) == 'API.bootstrap() already called'
+
+ def test_load_plugins(self):
+ """
+ Test the `ipalib.plugable.API.load_plugins` method.
+ """
+ (o, home) = self.new()
+ assert o.isdone('bootstrap') is False
+ assert o.isdone('load_plugins') is False
+ o.load_plugins()
+ assert o.isdone('bootstrap') is True
+ assert o.isdone('load_plugins') is True
+ e = raises(StandardError, o.load_plugins)
+ assert str(e) == 'API.load_plugins() already called'
diff --git a/tests/util.py b/tests/util.py
index 5656515c..cc761ce7 100644
--- a/tests/util.py
+++ b/tests/util.py
@@ -22,8 +22,71 @@ Common utility functions and classes for unit tests.
"""
import inspect
+import os
+from os import path
+import tempfile
+import shutil
from ipalib import errors
+
+class TempDir(object):
+ def __init__(self):
+ self.__path = tempfile.mkdtemp(prefix='ipa.tests.')
+ assert self.path == self.__path
+
+ def __get_path(self):
+ assert path.abspath(self.__path) == self.__path
+ assert self.__path.startswith('/tmp/ipa.tests.')
+ assert path.isdir(self.__path) and not path.islink(self.__path)
+ return self.__path
+ path = property(__get_path)
+
+ def rmtree(self):
+ if self.__path is not None:
+ shutil.rmtree(self.path)
+ self.__path = None
+
+ def makedirs(self, *parts):
+ d = self.join(*parts)
+ if not path.exists(d):
+ os.makedirs(d)
+ assert path.isdir(d) and not path.islink(d)
+ return d
+
+ def touch(self, *parts):
+ d = self.makedirs(*parts[:-1])
+ f = path.join(d, parts[-1])
+ assert not path.exists(f)
+ open(f, 'w').close()
+ assert path.isfile(f) and not path.islink(f)
+ return f
+
+ def write(self, content, *parts):
+ d = self.makedirs(*parts[:-1])
+ f = path.join(d, parts[-1])
+ assert not path.exists(f)
+ open(f, 'w').write(content)
+ assert path.isfile(f) and not path.islink(f)
+ return f
+
+ def join(self, *parts):
+ return path.join(self.path, *parts)
+
+ def __del__(self):
+ self.rmtree()
+
+
+class TempHome(TempDir):
+ def __init__(self):
+ super(TempHome, self).__init__()
+ self.__home = os.environ['HOME']
+ os.environ['HOME'] = self.path
+
+ def rmtree(self):
+ os.environ['HOME'] = self.__home
+ super(TempHome, self).rmtree()
+
+
class ExceptionNotRaised(Exception):
"""
Exception raised when an *expected* exception is *not* raised during a
generated by cgit (git 2.34.1) at 2025-08-31 01:47:40 +0000