IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

More CLI cleanup, got all basics working again

Browse Source
This commit is contained in:
Jason Gerard DeRose
2008-10-27 23:30:55 -06:00
parent 9b1e3f5946
commit 6e456cc749
4 changed files with 159 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
ipa
View File

@@ -28,10 +28,9 @@ The CLI functionality is implemented in ipalib/cli.py
import sys
from ipalib import api
from ipalib.cli import CLI
api.load_plugins()
if __name__ == '__main__':
cli = CLI(api,
(s.decode('utf-8') for s in sys.args[1:])
(s.decode('utf-8') for s in sys.argv[1:])
)
sys.exit(cli.run())

134
ipalib/cli.py
View File

@@ -218,9 +218,9 @@ class CLI(object):
self.argv = tuple(argv)
self.__done = set()
def run(self):
def run(self, init_only=False):
"""
Run a command (or attempt to at least).
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
@@ -245,6 +245,7 @@ class CLI(object):
"""
self.__doing('run')
self.finalize()
if self.api.env.mode == 'unit-test':
return
if len(self.cmd_argv) < 1:
self.print_commands()
@@ -255,10 +256,7 @@ class CLI(object):
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:])
)
return self.run_cmd(self[key])
def finalize(self):
"""
@@ -381,51 +379,38 @@ class CLI(object):
cmd.doc,
)
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]
def old_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 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:
self.run_interactive(cmd, kw)
kw = self.prompt_interactively(cmd, kw)
except KeyboardInterrupt:
return 0
except errors.RuleError, e:
# Now run the command
try:
ret = cmd(**kw)
if callable(cmd.output_for_cli):
cmd.output_for_cli(ret)
return 0
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:
if not (param.required or self.options.prompt_all):
continue
elif self.__not_interactive:
exit_error('Not enough arguments given')
default = param.get_default(**kw)
if default is None:
prompt = '%s: ' % param.cli_name
@@ -443,29 +428,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), KWCollector()
)
kw = kwc.__todict__()
try:
arg_kw = cmd.args_to_kw(*args)
@@ -492,10 +482,6 @@ class CLI(object):
parser.add_option(o)
return parser
def get_usage(self, cmd):
return ' '.join(self.get_usage_iter(cmd))
@@ -520,3 +506,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]

1
ipalib/constants.py
View File

@@ -53,6 +53,7 @@ DEFAULT_CONFIG = (
# Debugging:
('verbose', False),
('debug', False),
('mode', 'production'),
# ********************************************************
# The remaining keys are never set from the values here!

161
tests/test_ipalib/test_cli.py
View File

@@ -110,6 +110,7 @@ class test_CLI(ClassChecker):
frontend.Application,
backend.Backend,
)
api.env.mode = 'unit-test'
api.env.in_tree = True
o = self.cls(api, argv)
assert o.api is api
@@ -135,6 +136,92 @@ class test_CLI(ClassChecker):
assert o.api is api
assert o.argv == tuple(argv)
def test_run(self):
"""
Test the `ipalib.cli.CLI.run` method.
"""
self.check_cascade(
'run',
'finalize',
'load_plugins',
'bootstrap',
'parse_globals'
)
def test_finalize(self):
"""
Test the `ipalib.cli.CLI.finalize` method.
"""
self.check_cascade(
'finalize',
'load_plugins',
'bootstrap',
'parse_globals'
)
(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.load_plugins` method.
"""
self.check_cascade(
'load_plugins',
'bootstrap',
'parse_globals'
)
(o, api, home) = self.new()
assert api.isdone('load_plugins') is False
o.load_plugins()
assert api.isdone('load_plugins') is True
def test_bootstrap(self):
"""
Test the `ipalib.cli.CLI.bootstrap` method.
"""
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 the `ipalib.cli.CLI.parse_globals` method.
@@ -171,77 +258,3 @@ class test_CLI(ClassChecker):
assert o.cmd_argv == cmd_argv
e = raises(StandardError, o.parse_globals)
assert str(e) == 'CLI.parse_globals() already called'
def test_bootstrap(self):
"""
Test the `ipalib.cli.CLI.bootstrap` method.
"""
# Test with empty argv
(o, api, home) = self.new()
keys = tuple(api.env)
assert api.isdone('bootstrap') is False
assert o.isdone('parse_globals') is False
assert o.isdone('bootstrap') is False
o.bootstrap()
assert api.isdone('bootstrap') is True
assert o.isdone('parse_globals') is True
assert o.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_load_plugins(self):
"""
Test the `ipalib.cli.CLI.load_plugins` method.
"""
self.check_cascade(
'load_plugins',
'bootstrap',
'parse_globals'
)
(o, api, home) = self.new()
assert api.isdone('load_plugins') is False
o.load_plugins()
assert api.isdone('load_plugins') is True
def test_finalize(self):
"""
Test the `ipalib.cli.CLI.finalize` method.
"""
self.check_cascade(
'finalize',
'load_plugins',
'bootstrap',
'parse_globals'
)
(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)