---
matahari.spec.in | 9
src/python/CMakeLists.txt | 14
src/python/matahari/interpreter/__init__.py | 3
src/python/matahari/interpreter/command.py | 976 --------------------
src/python/matahari/interpreter/interpreter.py | 125 ---
src/python/matahari/interpreter/mode.py | 80 --
src/python/matahari/shell.py | 12
src/python/matahari/shell/__init__.py | 12
src/python/matahari/shell/interpreter/__init__.py | 3
src/python/matahari/shell/interpreter/command.py | 976 ++++++++++++++++++++
.../matahari/shell/interpreter/interpreter.py | 125 +++
src/python/matahari/shell/interpreter/mode.py | 80 ++
src/python/matahari/shell/mode.py | 220 +++++
src/python/matahari/shell/types.py | 312 ++++++
src/python/matahari/shellmode.py | 220 -----
src/python/matahari/shelltypes.py | 312 ------
16 files changed, 1741 insertions(+), 1738 deletions(-)
delete mode 100644 src/python/matahari/interpreter/__init__.py
delete mode 100644 src/python/matahari/interpreter/command.py
delete mode 100644 src/python/matahari/interpreter/interpreter.py
delete mode 100644 src/python/matahari/interpreter/mode.py
delete mode 100644 src/python/matahari/shell.py
create mode 100644 src/python/matahari/shell/__init__.py
create mode 100644 src/python/matahari/shell/interpreter/__init__.py
create mode 100644 src/python/matahari/shell/interpreter/command.py
create mode 100644 src/python/matahari/shell/interpreter/interpreter.py
create mode 100644 src/python/matahari/shell/interpreter/mode.py
create mode 100644 src/python/matahari/shell/mode.py
create mode 100644 src/python/matahari/shell/types.py
delete mode 100644 src/python/matahari/shellmode.py
delete mode 100644 src/python/matahari/shelltypes.py
diff --git a/matahari.spec.in b/matahari.spec.in
index c26261f..86af42c 100644
--- a/matahari.spec.in
+++ b/matahari.spec.in
@@ -538,14 +538,17 @@ test "x%{buildroot}" != "x" && rm -rf
%{buildroot}
%{python_sitelib}/matahari/*.py
%{python_sitelib}/matahari/*.pyc
%{python_sitelib}/matahari/*.pyo
-%{python_sitelib}/matahari/interpreter/*.py
-%{python_sitelib}/matahari/interpreter/*.pyc
-%{python_sitelib}/matahari/interpreter/*.pyo
%files shell
%defattr(644, root, root, 755)
%doc AUTHORS COPYING
%{_bindir}/mhsh
+%{python_sitelib}/matahari/shell/*.py
+%{python_sitelib}/matahari/shell/*.pyc
+%{python_sitelib}/matahari/shell/*.pyo
+%{python_sitelib}/matahari/shell/interpreter/*.py
+%{python_sitelib}/matahari/shell/interpreter/*.pyc
+%{python_sitelib}/matahari/shell/interpreter/*.pyo
%files network
%defattr(644, root, root, 755)
diff --git a/src/python/CMakeLists.txt b/src/python/CMakeLists.txt
index 2468ced..0a8a428 100644
--- a/src/python/CMakeLists.txt
+++ b/src/python/CMakeLists.txt
@@ -26,13 +26,13 @@ if(WITH-QMF)
py_module(rpc_server matahari)
py_module(core matahari)
py_module(api matahari)
- py_module(shell matahari)
- py_module(shellmode matahari)
- py_module(shelltypes matahari)
- py_module(__init__ matahari/interpreter)
- py_module(interpreter matahari/interpreter)
- py_module(mode matahari/interpreter)
- py_module(command matahari/interpreter)
+ py_module(__init__ matahari/shell)
+ py_module(mode matahari/shell)
+ py_module(types matahari/shell)
+ py_module(__init__ matahari/shell/interpreter)
+ py_module(interpreter matahari/shell/interpreter)
+ py_module(mode matahari/shell/interpreter)
+ py_module(command matahari/shell/interpreter)
install(PROGRAMS ${CMAKE_CURRENT_SOURCE_DIR}/mhsh
DESTINATION bin)
diff --git a/src/python/matahari/interpreter/__init__.py
b/src/python/matahari/interpreter/__init__.py
deleted file mode 100644
index 96af66f..0000000
--- a/src/python/matahari/interpreter/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from interpreter import Interpreter
-from mode import Mode
-from command import Command, InvalidArgumentException
diff --git a/src/python/matahari/interpreter/command.py
b/src/python/matahari/interpreter/command.py
deleted file mode 100644
index f857196..0000000
--- a/src/python/matahari/interpreter/command.py
+++ /dev/null
@@ -1,976 +0,0 @@
-"""
-Module for defining the syntax of CLI commands.
-
-To define a command, create a handler function and wrap it with the Command
-decorator, like so:
-
- @Command('test', 'PARAM')
- def test(kw_test, param):
- '''
- A test command
-
- Prints the parameter passed in.
- '''
- assert kw_test == 'test'
- print param
-
-The parameters to the Command constructor are the arguments to the command.
-Arguments that are not in all caps are fixed keywords.
-
-Arguments in all caps (e.g. 'PARAM') are variable parameters. Additionally,
-an argument that is a function will be treated as a type conversion and
-validation function for a variable parameter. For example, pass the builtin
-int class to convert the argument to an integer. The function should throw
-a ValueError exception if the argument cannot be converted. If the function
-has an attribute named "complete", this will be called to provide suggested
-values for tab-completion. The complete method is passed the current
-fragment that the user is trying to tab-complete, and should return a list
-of matching allowable values.
-
-A group of optional parameters can be passed as a tuple. For example:
-
- @Command('test2', ('optional-keyword', 'PARAM'),
'MANDATORY')
- def test2(kw_test, (kw_optkw, param), mandatory):
- '''A test command with optional parameters'''
- if kw_optkw is not None:
- print 'Optional:', param
- print 'Mandatory:', mandatory
-
-Note that if the optional parameters are not supplied, they will each have
-the value None when passed to the handler function. This is done (instead of
-passing an empty list) so that tuple unpacking can be used as shown in the
-example. If a validation function has an attribute named "default", this
-default value will be used instead of None. If the default attribute is
-callable, it will be called to calculate the default.
-
-A variable argument list can be implemented by passing a list as the final
-parameter, thus:
-
- @Command('test3', ['PARAMS'])
- def test3(kw_test, *params):
- '''A test command with variable arguments'''
- print 'Params:' params
-
-A least one argument must be supplied for a variable argument list. To allow
-a list with zero items, make it optional by wrapping it in a tuple. In the case
-of variable argument lists, the default return value is an empty list (rather
-than None as for other parameters).
-
-The docstring of the command handler function will be used to display help
-about the command to the user on request.
-"""
-
-
-import types
-from itertools import *
-
-
-class Command(object):
- """
- Decorator to define the argument list for a command handler.
-
- Arguments may be keywords (lower-case strings), string parameters
- (upper-case strings), or parameters of different types (functions that
- validate and/or convert the parameters from strings).
-
- A series of arguments can be made optional by enclosing them in a tuple,
- or repeated by enclosing in a list.
- """
-
- def __init__(self, name, *args):
- """Initialise with a list of command arguments."""
- self.name = Keyword(name)
- self.args = [Argument(a) for a in args]
-
- def __call__(self, func):
- """Decorate the command handler."""
- return CommandHandler(func, self.name, self.args)
-
-
-class InvalidArgumentException(Exception):
- """
- An Exception raised when the value of an argument passed by the user is not
- valid in the current context.
- """
- pass
-
-class InvalidCommandException(Exception):
- """
- An Exception raised when a command executed by the user is invalid.
- """
-
- def __init__(self, line, candidates=[], message=None):
- self.line = line
- self.candidates = candidates
- if message is None:
- message = 'Invalid Command: "%s"' % self.line
-
- if self.candidates:
- message += ('\nCandidates are:\n' +
- '\n'.join(' ' + str(c) for c, e in
self.candidates))
-
- super(InvalidCommandException, self).__init__(message)
-
-
-class CommandHandler(object):
- """A handler for a command."""
-
- def __init__(self, func, name, args):
- """Initialise with a handler function and a list of
arguments."""
- self.do = func
- self.name = str(name)
- self.args = [name] + args
- self.arg_graph = ArgGraph(self.args)
- self.__name__ = self.do.__name__
- self.__doc__ = self.help()
-
- def help(self):
- """Return the help string for the command."""
- lines = [' > %s\n' % (str(self),)]
- doc = self._trimdoc(self.do.__doc__)
- if doc:
- lines.append(doc + '\n')
- return '\n'.join(lines)
-
- def complete(self, text, line, begidx, endidx):
- """Return a list of tab-completion options for the
command."""
- values = line.split()
- if values[-1] != text:
- values.append(text)
-
- match = lambda (a, v): a and a.complete(v) or []
-
- def completions(arglist):
- argvals = takewhile(lambda (a, v): v is not None,
- izip_longest(arglist, values, fillvalue=None))
- for m in imap(match, argvals):
- if not m:
- return []
- return m
-
- candidates = chain.from_iterable(imap(completions, self.arg_graph))
-
- return list(set(candidates))
-
- def __get__(self, obj, objtype=None):
- # Allow the handler to be bound to objects as a method
- boundfunc = types.MethodType(self.do, obj, objtype)
- return CommandHandler(boundfunc, self.args[0], self.args[1:])
-
- def __call__(self, line):
- """Do the command."""
- vals = [self.name] + line.split()
- args = tuple(Argument.collate(self.args, vals))
-
- if len(args) < len(self.args):
- miss = ' '.join(str(a) for a in self.args[len(args):])
- raise InvalidCommandException(line,
- message="Missing arguments: '%s'" % miss)
- if vals:
- extra = ' '.join(vals)
- raise InvalidCommandException(line,
- message="Excess arguments: '%s'" % extra)
-
- return self.do(*args)
-
- def __repr__(self):
- return '@Command(%s)' % ', '.join(repr(a) for a in self.args)
-
- def __str__(self):
- return ' '.join(str(a) for a in self.args)
-
- def __add__(self, other):
- if other is None:
- return self
- return CommandGroupHandler(self.name, self, other)
-
- def __radd__(self, other):
- if other is None:
- return self
- return CommandGroupHandler(self.name, other, self)
-
- @staticmethod
- def _trimdoc(docstring):
- """Trim whitespace from a docstring."""
- if not docstring:
- return ''
- # Convert tabs to spaces (following the normal Python rules)
- # and split into a list of lines:
- lines = docstring.expandtabs().splitlines()
- # Determine minimum indentation (first line doesn't count):
- nonblank = lambda l: l and not l.isspace()
- indents = [len(l) - len(l.lstrip()) for l in lines[1:] if nonblank(l)]
- indent = indents and reduce(min, indents) or 0
- # Remove indentation (first line is special):
- trimmed = [lines[0].strip()] + [l[indent:].rstrip() for l in lines[1:]]
- # Strip off trailing and leading blank lines:
- while trimmed and not trimmed[-1]:
- trimmed.pop()
- while trimmed and not trimmed[0]:
- trimmed.pop(0)
- # Return a single string:
- return '\n'.join(trimmed)
-
-
-class CommandGroupHandler(object):
- """A handler for a group of commands with the same
name."""
-
- def __init__(self, *args):
- """
- Initialise with an optional command name and a series of command
- handlers.
- """
- if isinstance(args[0], basestring):
- self.name = args[0]
- args = args[1:]
- else:
- self.name = args[0].name
- self.cmds = []
- self.__doc__ = None
- for c in args:
- self += c
- self.__doc__ = self.help()
-
- def help(self):
- """Return the help string for the command."""
- return '\n\n'.join(c.help() for c in self.cmds)
-
- def complete(self, text, line, begidx, endidx):
- """Return a list of tab-completion options for the
command."""
- completions = lambda c: c.complete(text, line, begidx, endidx)
- return list(set().union(*map(completions, self.cmds)))
-
- def _check_add(self, other):
- if not isinstance(other, CommandHandler):
- raise TypeError("%s is not a CommandHandler" % type(other))
- if other.name != self.name:
- raise ValueError('Command name "%s" does not match group
"%s"' %
- (other.name, self.name))
-
- def __add__(self, other):
- """Create a new group including an additional command or
group."""
- if other is None:
- return self
- if isinstance(other, CommandGroupHandler):
- cmds = other.cmds
- if other.name != self.name:
- raise ValueError('Command names "%s" and "%s" do
not match' %
- (self.name, other.name))
- else:
- self._check_add(other)
- cmds = [other]
- return CommandGroupHandler(self.name, *(self.cmds + cmds))
-
- def __iadd__(self, other):
- """Add a new command to the group."""
- self._check_add(other)
- self.cmds.append(other)
- if self.__doc__ is not None:
- self.__doc__ = self.help()
- return self
-
- def __repr__(self):
- return 'CommandGroupHandler(%s)' % repr(self.cmds)
-
- def __str__(self):
- return '\n'.join(' ' + str(c) for c in self.cmds)
-
- def __call__(self, line):
- """Do the command."""
- failures = []
-
- for c in self.cmds:
- try:
- return c(line)
- except InvalidCommandException, e:
- failures.extend(e.candidates or [(c, None)])
- except InvalidArgumentException, e:
- failures.append((c, e))
-
- line = ' '.join((self.name, line))
- raise InvalidCommandException(line.strip(), candidates=failures)
-
-
-class ArgGraph(object):
- """A representation of the allowable sequences of command arguments in
the
- form of a directed acyclic graph."""
-
- def __init__(self, args):
- """Initialise with a list of arguments"""
- self.edges = {}
- roots, leaves = self._addargs(args)
- self.first = roots[0]
-
- def __getitem__(self, key):
- if key is None:
- return self.first
- return self.edges[key]
-
- def __iter__(self):
- """
- Return a generator for all of the allowed sequences of arguments.
- """
- return self._getpaths()
-
- def _getpaths(self, visited=None):
- """
- Return a generator for all of the allowed sequences of agruments
- beginning with the given subpath.
- """
- children = self[visited and visited[-1]]
-
- isleaf = lambda n: n not in self.edges
- pathto = lambda n: (visited or []) + [n]
-
- # Paths that end with the start node's children
- leaf_paths = imap(lambda leaf: pathto(leaf),
- ifilter(isleaf, children))
-
- # Treat each Argument as an iterator and chain their output to pick up
- # internal members of e.g. RepeatedArguments
- arg_chains = imap(chain.from_iterable, leaf_paths)
-
- getpaths = lambda n: self._getpaths(pathto(n))
- nonleaf_children = ifilterfalse(isleaf, children)
-
- # Paths that are recursively obtained via non-leaf children
- child_arg_chains = chain.from_iterable(imap(getpaths,
- nonleaf_children))
-
- return chain(arg_chains, child_arg_chains)
-
- def _addargs(self, args):
- """Add a list of arguments to the Graph"""
- roots_found = False
- roots = []
- leaves = []
-
- optional = lambda n: isinstance(n, OptionalArguments)
-
- def handle_opt_args(opt_args):
- r, l = self._addargs(opt_args)
- self._addedges(leaves, *r)
- if not roots_found:
- roots.extend(r)
- leaves.extend(l)
-
- def handle_arg(arg):
- self._addedges(leaves, arg)
- if not roots_found:
- roots.append(arg)
- leaves[:] = [arg]
-
- for a in args:
- if optional(a):
- handle_opt_args(a)
- else:
- handle_arg(a)
- roots_found = True
-
- return roots, leaves
-
- def _addedges(self, srcs, *dests):
- for s in srcs:
- l = self.edges.get(s, [])
- l.extend(dests)
- self.edges[s] = l
-
-
-class Argument(object):
- """The definition of an argument to a command"""
-
- def __new__(cls, arg):
- """Create an argument of the correct type from its
definition."""
- if cls != Argument:
- return super(Argument, cls).__new__(cls)
-
- if isinstance(arg, tuple):
- argclass = OptionalArguments
- elif isinstance(arg, list):
- argclass = RepeatedArguments
- elif callable(arg):
- argclass = Parameter
- elif isinstance(arg, basestring):
- if arg.upper() == arg:
- argclass = Parameter
- else:
- argclass = Keyword
- else:
- return arg
- return argclass(arg)
-
- @staticmethod
- def collate(args, values):
- """
- Match values to arguments and return an argument list suitable for
- passing to the handler function.
- """
- arglist = chain.from_iterable(imap(lambda a: a.match(values), args))
- return takewhile(lambda p: p is not None, arglist)
-
- def default(self):
- return None
-
- def __iter__(self):
- yield self
-
-
-class Parameter(Argument):
- """A parameter passed to the command by the user"""
- def __init__(self, param):
- self.param = param
-
- def __repr__(self):
- if callable(self.param):
- if hasattr(self.param, '__name__'):
- return self.param.__name__
- else:
- return type(self.param).__name__
- return str(self.param)
-
- def __str__(self):
- return repr(self).upper()
-
- def __call__(self, arg):
- if callable(self.param):
- try:
- return self.param(arg)
- except ValueError, e:
- raise InvalidArgumentException('Invalid argument value (%s)' %
e)
- return arg
-
- def default(self):
- if hasattr(self.param, 'default'):
- if callable(self.param.default):
- return self.param.default()
- else:
- return self.param.default
- return None
-
- def complete(self, value):
- if hasattr(self.param, 'complete') and callable(self.param.complete):
- return self.param.complete(value)
- try:
- if value:
- parsed = self(value)
- return [value and (value + ' ')]
- except InvalidArgumentException:
- return []
-
- def match(self, values):
- if not values:
- yield None
- return
- yield self(values.pop(0))
-
-
-class Keyword(Argument):
- """A fixed keyword that must be present in the
command"""
- def __init__(self, kw):
- self.kw = kw
-
- def __str__(self):
- return self.kw
-
- def __eq__(self, other):
- return self.kw == other
-
- def __ne__(self, other):
- return self.kw != other
-
- def complete(self, value):
- if self.kw.startswith(value):
- return [self.kw + ' ']
- else:
- return []
-
- def match(self, values):
- if not values:
- yield None
- return
- val = values[0]
- if self != val:
- raise InvalidArgumentException("Invalid keyword '%s'" %
val)
- yield values.pop(0)
-
-
-class ArgumentList(Argument):
- """Abstract base class for lists of arguments"""
-
- def __init__(self, args):
- self.args = [Argument(a) for a in args]
-
- def __iter__(self):
- return iter(self.args)
-
- def __repr__(self):
- return ', '.join(repr(a) for a in self.args)
-
- def __str__(self):
- return ' '.join(str(a) for a in self.args)
-
-
-class OptionalArguments(ArgumentList):
- """A list of arguments that may be omitted when the command is
called"""
-
- def __repr__(self):
- contents = ArgumentList.__repr__(self)
- if len(self.args) == 1:
- contents += ','
- return '(%s)' % contents
-
- def __str__(self):
- return '(%s)' % ArgumentList.__str__(self)
-
- def default(self):
- return [a.default() for a in self.args]
-
- def match(self, values):
- args = self.default()
- if values:
- values_copy = list(values)
- try:
- args = list(self.collate(self.args, values_copy))
- values[:] = values_copy
- except InvalidArgumentException:
- pass
- yield args
-
-
-class RepeatedArguments(ArgumentList):
- """A list of arguments that may be repeated"""
-
- def __iter__(self):
- return cycle(self.args)
-
- def __repr__(self):
- return '[%s]' % ArgumentList.__repr__(self)
-
- def __str__(self):
- return '[%s]' % ArgumentList.__str__(self)
-
- def match(self, values):
- if not values:
- raise InvalidArgumentException("Empty variadic argument list")
-
- values_remain = lambda c: values
-
- iterations = takewhile(values_remain,
- starmap(self.collate,
- repeat((self.args, values))))
- return chain.from_iterable(iterations)
-
- def default(self):
- return []
-
-
-import unittest
-
-class CommandTest(unittest.TestCase):
- class CallCounter(object):
- def __init__(self, func):
- self.func = func
- self.__name__ = func.__name__
- self.calls = 0
-
- def __call__(self, *args, **kwargs):
- self.calls += 1
- return self.func(*args, **kwargs)
-
-
- def setUp(self):
- def nullCmd(*args, **kwargs):
- self.fail('Command handler erroneously called')
- self.nullCmd = nullCmd
-
- def test_simple(self):
- @self.CallCounter
- def foo(arg):
- self.assertEqual(arg, 'foo')
- handler = Command('foo')(foo)
- handler('')
- self.assertEqual(foo.calls, 1)
-
- def test_simple_bad(self):
- handler = Command('foo')(self.nullCmd)
- self.assertRaises(InvalidCommandException, handler, 'bar')
-
- def test_param(self):
- @self.CallCounter
- def foo(kwfoo, param):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- handler = Command('foo', 'PARAM')(foo)
- handler('bar')
- self.assertEqual(foo.calls, 1)
-
- def test_param_missing(self):
- handler = Command('foo', 'PARAM')(self.nullCmd)
- self.assertRaises(InvalidCommandException, handler, '')
-
- def test_param_extra(self):
- handler = Command('foo')(self.nullCmd)
- self.assertRaises(InvalidCommandException, handler, 'bar')
-
- def test_param_validator(self):
- @self.CallCounter
- def foo(kwfoo, intarg):
- self.assertEqual(kwfoo, 'foo')
- self.assertIsInstance(intarg, int)
- self.assertEqual(intarg, 42)
- handler = Command('foo', int)(foo)
- handler('42')
- self.assertEqual(foo.calls, 1)
-
- def test_param_validator_invalid(self):
- @self.CallCounter
- def foo(kwfoo, intarg):
- self.fail('Command handler called')
- handler = Command('foo', int)(foo)
- self.assertRaises(InvalidArgumentException, handler, 'bar')
-
- def test_opt_args(self):
- @self.CallCounter
- def foo(kwfoo, (kwbar, param1), param2):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(kwbar, 'bar')
- self.assertEqual(param1, 'baz')
- self.assertEqual(param2, 'blarg')
- handler = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(foo)
- handler('bar baz blarg')
- self.assertEqual(foo.calls, 1)
-
- def test_opt_args_missing(self):
- @self.CallCounter
- def foo(kwfoo, optargs, param2):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(optargs, [None, None])
- self.assertEqual(param2, 'wibble')
- handler = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(foo)
- handler('wibble')
- self.assertEqual(foo.calls, 1)
-
- def test_opt_args_partial(self):
- handler = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(self.nullCmd)
- self.assertRaises(InvalidCommandException, handler, 'bar')
-
- def test_opt_args_end_missing(self):
- @self.CallCounter
- def foo(kwfoo, optargs):
- self.assertEqual(kwfoo, 'foo')
- self.assertSequenceEqual(optargs, [None])
- handler = Command('foo', ('bar',))(foo)
- handler('')
- self.assertEqual(foo.calls, 1)
-
- def test_opt_kw(self):
- @self.CallCounter
- def foo(kwfoo, (kwbar,), param):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(kwbar, 'bar')
- self.assertEqual(param, 'baz')
- handler = Command('foo', ('bar',), 'PARAM')(foo)
- handler('bar baz')
- self.assertEqual(foo.calls, 1)
-
- def test_opt_kw_missing(self):
- @self.CallCounter
- def foo(kwfoo, (kwbar,), param):
- self.assertEqual(kwfoo, 'foo')
- self.assertIs(kwbar, None)
- self.assertEqual(param, 'quux')
- handler = Command('foo', ('bar',), 'PARAM')(foo)
- handler('quux')
- self.assertEqual(foo.calls, 1)
-
- def test_opt_kw_partial(self):
- handler = Command('foo', ('bar',),
'PARAM')(self.nullCmd)
- self.assertRaises(InvalidCommandException, handler, 'bar')
-
- def test_opt_kw_missing(self):
- @self.CallCounter
- def foo(kwfoo, (kwbar,), param):
- self.assertEqual(kwfoo, 'foo')
- self.assertIs(kwbar, None)
- self.assertEqual(param, 'quux')
- handler = Command('foo', ('bar',), 'PARAM')(foo)
- handler('quux')
- self.assertEqual(foo.calls, 1)
-
- def test_list(self):
- @self.CallCounter
- def foo(kwfoo, param, *args):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- self.assertSequenceEqual(args, ['baz', 'blarg',
'wibble'])
- handler = Command('foo', 'bar', ['PARAMS'])(foo)
- handler('bar baz blarg wibble')
- self.assertEqual(foo.calls, 1)
-
- def test_list_single(self):
- @self.CallCounter
- def foo(kwfoo, param, *args):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- self.assertSequenceEqual(args, ['baz'])
- handler = Command('foo', 'bar', ['PARAMS'])(foo)
- handler('bar baz')
- self.assertEqual(foo.calls, 1)
-
- def test_list_empty(self):
- handler = Command('foo', 'bar',
['PARAMS'])(self.nullCmd)
- self.assertRaises(InvalidArgumentException, handler, 'bar')
-
- def test_list_optional(self):
- @self.CallCounter
- def foo(kwfoo, param, args):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- self.assertSequenceEqual(args, [42, 43])
- handler = Command('foo', 'bar', ([int],))(foo)
- handler('bar 42 43')
- self.assertEqual(foo.calls, 1)
-
- def test_list_optional_empty(self):
- @self.CallCounter
- def foo(kwfoo, param, args):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- self.assertEqual(args[0], [])
- handler = Command('foo', 'bar', ([int],))(foo)
- handler('bar')
- self.assertEqual(foo.calls, 1)
-
- def test_list_optional_kw_err(self):
- handler = Command('foo', ('bar',
['PARAMS']))(self.nullCmd)
- self.assertRaises(InvalidCommandException, handler, 'bar')
-
- def test_default_param(self):
- class Answer(int):
- default = 42
-
- @self.CallCounter
- def foo(kwfoo, param, args):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- self.assertEqual(args[0], 42)
- handler = Command('foo', 'bar', (Answer,))(foo)
- handler('bar')
- self.assertEqual(foo.calls, 1)
-
- def test_default_param_func(self):
- class Answer(int):
- @staticmethod
- def default():
- return 42
-
- @self.CallCounter
- def foo(kwfoo, param, args):
- self.assertEqual(kwfoo, 'foo')
- self.assertEqual(param, 'bar')
- self.assertEqual(args[0], 42)
- handler = Command('foo', 'bar', (Answer,))(foo)
- handler('bar')
- self.assertEqual(foo.calls, 1)
-
-
-class CompletionTest(unittest.TestCase):
- def handler(self, *args):
- @Command(*args)
- def nullCmd(*args, **kwargs):
- self.fail('Command handler called')
- return nullCmd
-
- def assertCompletion(self, handler, line, expected):
- actual = self.complete(handler, line)
- err = '\n' + '\n'.join(('Command: "%s"' %
handler,
- 'Typed: "%s"' % line,
- 'Completions: %s' % actual,
- 'Expected: %s' % expected))
- self.assertItemsEqual(actual, expected, err)
-
- def complete(self, handler, line):
- parts = line.rpartition(' ')
- text = parts[2]
- begidx = len(parts[0]) + len(parts[1])
- endidx = len(line)
- return handler.complete(text, line, begidx, endidx)
-
- def test_kw(self):
- h = self.handler('foo', 'bar')
- self.assertCompletion(h, 'foo ', ['bar '])
- self.assertCompletion(h, 'foo b', ['bar '])
- self.assertCompletion(h, 'foo bar ', [])
-
- def test_kw_space(self):
- h = self.handler('foo', 'bar')
- self.assertCompletion(h, 'foo', ['foo '])
- self.assertCompletion(h, 'foo bar', ['bar '])
-
- def test_multiple_kw(self):
- h = self.handler('foo', 'bar', 'baz')
- self.assertCompletion(h, 'foo b', ['bar '])
- self.assertCompletion(h, 'foo bar ', ['baz '])
-
- def test_param(self):
- h = self.handler('foo', 'PARAM')
- self.assertCompletion(h, 'foo ', [''])
- self.assertCompletion(h, 'foo b', ['b '])
- self.assertCompletion(h, 'foo bar ', [])
-
- def test_param_validator(self):
- h = self.handler('foo', int)
- self.assertCompletion(h, 'foo ', [''])
- self.assertCompletion(h, 'foo b', [])
- self.assertCompletion(h, 'foo 1', ['1 '])
- self.assertCompletion(h, 'foo 123 ', [])
-
- def test_param_suggestor(self):
- candidates = ['bar ', 'baz ', 'blarg ']
- class Param(object):
- def __call__(self):
- pass
- def complete(self, arg):
- return [a for a in candidates if a.startswith(arg)]
-
- h = self.handler('foo', Param())
- self.assertCompletion(h, 'foo ', candidates)
- self.assertCompletion(h, 'foo b', candidates)
- self.assertCompletion(h, 'foo ba', ['bar ', 'baz '])
- self.assertCompletion(h, 'foo bar', ['bar '])
- self.assertCompletion(h, 'foo bar ', [])
- self.assertCompletion(h, 'foo quux', [])
-
- def test_multiple_param(self):
- h = self.handler('foo', 'PARAM', 'PARAM2')
- self.assertCompletion(h, 'foo ', [''])
- self.assertCompletion(h, 'foo b', ['b '])
- self.assertCompletion(h, 'foo bar ', [''])
- self.assertCompletion(h, 'foo bar baz', ['baz '])
- self.assertCompletion(h, 'foo bar baz ', [])
-
- def test_opt_args(self):
- h = self.handler('foo', ('bar', 'PARAM1'),
'PARAM2')
- self.assertCompletion(h, 'foo ', ['bar ', ''])
- self.assertCompletion(h, 'foo b', ['bar ', 'b '])
- self.assertCompletion(h, 'foo bar ', [''])
- self.assertCompletion(h, 'foo quux', ['quux '])
- self.assertCompletion(h, 'foo quux ', [])
-
- def test_multiple_opt_args(self):
- h = self.handler('foo', ('bar', 'PARAM1'),
('baz', 'PARAM2'))
- self.assertCompletion(h, 'foo', ['foo '])
- self.assertCompletion(h, 'foo ', ['bar ', 'baz '])
- self.assertCompletion(h, 'foo b', ['bar ', 'baz '])
- self.assertCompletion(h, 'foo bar ', [''])
- self.assertCompletion(h, 'foo baz ', [''])
- self.assertCompletion(h, 'foo baz quux ', [])
-
- def test_opt_kw(self):
- h = self.handler('foo', ('bar',), 'PARAM')
- self.assertCompletion(h, 'foo ', ['bar ', ''])
- self.assertCompletion(h, 'foo b', ['bar ', 'b '])
- self.assertCompletion(h, 'foo bar ', [''])
- self.assertCompletion(h, 'foo bar quux', ['quux '])
- self.assertCompletion(h, 'foo bar quux ', [])
- self.assertCompletion(h, 'foo quux', ['quux '])
- self.assertCompletion(h, 'foo quux ', [])
-
- def test_multiple_opt_kw_first(self):
- h = self.handler('foo', ('bar',), ('baz',))
- self.assertCompletion(h, 'foo', ['foo '])
- self.assertCompletion(h, 'foo ', ['bar ', 'baz '])
- self.assertCompletion(h, 'foo b', ['bar ', 'baz '])
- self.assertCompletion(h, 'foo bar ', ['baz '])
- self.assertCompletion(h, 'foo baz ', [])
- self.assertCompletion(h, 'foo quux ', [])
- self.assertCompletion(h, 'foo bar quux ', [])
-
- def test_multiple_opt_kw(self):
- h = self.handler('foo', 'wibble', ('bar',),
('baz',))
- self.assertCompletion(h, 'foo', ['foo '])
- self.assertCompletion(h, 'foo ', ['wibble '])
- self.assertCompletion(h, 'foo wibble ', ['bar ', 'baz
'])
- self.assertCompletion(h, 'foo wibble b', ['bar ', 'baz
'])
- self.assertCompletion(h, 'foo wibble bar ', ['baz '])
- self.assertCompletion(h, 'foo wibble baz ', [])
- self.assertCompletion(h, 'foo wibble quux ', [])
- self.assertCompletion(h, 'foo wibble bar quux ', [])
-
- def test_list(self):
- h = self.handler('foo', 'bar', ['PARAMS'])
- self.assertCompletion(h, 'foo ', ['bar '])
- self.assertCompletion(h, 'foo bar', ['bar '])
- self.assertCompletion(h, 'foo bar ', [''])
- self.assertCompletion(h, 'foo bar baz', ['baz '])
- self.assertCompletion(h, 'foo bar baz ', [''])
- self.assertCompletion(h, 'foo bar baz blarg', ['blarg '])
- self.assertCompletion(h, 'foo bar baz blarg ', [''])
- self.assertCompletion(h, 'foo bar baz blarg wibble', ['wibble
'])
- self.assertCompletion(h, 'foo bar baz blarg wibble ', [''])
-
- def test_list_optional(self):
- h = self.handler('foo', 'bar', (['PARAMS'],))
- self.assertCompletion(h, 'foo ', ['bar '])
- self.assertCompletion(h, 'foo bar', ['bar '])
- self.assertCompletion(h, 'foo bar ', [''])
- self.assertCompletion(h, 'foo bar baz', ['baz '])
- self.assertCompletion(h, 'foo bar baz ', [''])
- self.assertCompletion(h, 'foo bar baz blarg', ['blarg '])
- self.assertCompletion(h, 'foo bar baz blarg ', [''])
- self.assertCompletion(h, 'foo bar baz blarg wibble', ['wibble
'])
- self.assertCompletion(h, 'foo bar baz blarg wibble ', [''])
-
- def test_list_kw_optional(self):
- h = self.handler('foo', 'bar', ('baz',
['PARAMS'],))
- self.assertCompletion(h, 'foo ', ['bar '])
- self.assertCompletion(h, 'foo bar', ['bar '])
- self.assertCompletion(h, 'foo bar ', ['baz '])
- self.assertCompletion(h, 'foo bar baz', ['baz '])
- self.assertCompletion(h, 'foo bar baz ', [''])
- self.assertCompletion(h, 'foo bar baz blarg', ['blarg '])
- self.assertCompletion(h, 'foo bar baz blarg ', [''])
- self.assertCompletion(h, 'foo bar baz blarg wibble', ['wibble
'])
- self.assertCompletion(h, 'foo bar baz blarg wibble ', [''])
-
-
-class SyntaxTest(unittest.TestCase):
- def setUp(self):
- def nullCmd(*args, **kwargs):
- self.fail('Command handler called')
- self.nullCmd = nullCmd
-
- def test_simple(self):
- foo = Command('foo')(self.nullCmd)
- self.assertEqual(str(foo), 'foo')
-
- def test_param(self):
- foo = Command('foo', 'PARAM')(self.nullCmd)
- self.assertEqual(str(foo), 'foo PARAM')
-
- def test_param_validator(self):
- foo = Command('foo', int)(self.nullCmd)
- self.assertEqual(str(foo), 'foo INT')
-
- def test_opt_args(self):
- foo = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(self.nullCmd)
- self.assertEqual(str(foo), 'foo (bar PARAM1) PARAM2')
-
- def test_opt_kw(self):
- foo = Command('foo', ('bar',), 'PARAM')(self.nullCmd)
- self.assertEqual(str(foo), 'foo (bar) PARAM')
-
- def test_list(self):
- foo = Command('foo', ['PARAMS'])(self.nullCmd)
- self.assertEqual(str(foo), 'foo [PARAMS]')
-
- def test_opt_list(self):
- foo = Command('foo', (['PARAMS'],))(self.nullCmd)
- self.assertEqual(str(foo), 'foo ([PARAMS])')
-
-def suite():
- suite = unittest.TestSuite()
- suite.addTest(CommandTest)
- suite.addTest(CompletionTest)
- suite.addTest(SyntaxTest)
- return suite
diff --git a/src/python/matahari/interpreter/interpreter.py
b/src/python/matahari/interpreter/interpreter.py
deleted file mode 100644
index 9eb1dbe..0000000
--- a/src/python/matahari/interpreter/interpreter.py
+++ /dev/null
@@ -1,125 +0,0 @@
-"""
-Module defining a modal command interpreter.
-
-The commands supported by the interpreter are supplied by the current mode,
-which can be changed by passing a new mode to the set_mode() method.
-"""
-
-import cmd
-
-import mode
-from command import InvalidCommandException, InvalidArgumentException
-
-
-class Interpreter(cmd.Cmd):
- """A modal command interpreter."""
-
- def __init__(self, prompt, mode=mode.Mode(), debug=False):
- """Initialise the interpreter in the initial
mode."""
- cmd.Cmd.__init__(self)
- self.name = prompt
- self.debug = debug
- self.mode = None
- self.set_mode(mode)
- self.doc_header = 'Commands:'
- self.misc_header = 'Help topics:'
- self.undoc_header = ''
-
- def set_mode(self, mode):
- """Set the shell mode."""
- if self.mode is not None:
- self.mode.deactivate(self)
- self.mode = mode
- self.mode.activate(self)
- custom_prompt = mode.prompt()
- self.prompt = self.name + (custom_prompt and ' ') + custom_prompt +
'> '
-
- def completenames(self, *args):
- # Tab-complete should move to the next argument
- return [s + ' ' for s in cmd.Cmd.completenames(self, *args)]
-
- def complete_help(self, *args):
- # Don't suggest multiple arguments
- if len(args[1].split()) > (args[1][-1] == ' ' and 1 or 2):
- return []
- # Explicitly call superclass's completenames() method, to avoid adding
- # spaces (and thus duplicate output when combined with the topic list)
- commands = set(cmd.Cmd.completenames(self, *args))
- topics = set(a[5:] for a in self.get_names()
- if a.startswith('help_' + args[0]))
- return list(commands | topics)
-
- def emptyline(self):
- # Don't do anything for an empty command
- pass
-
- def do_EOF(self, line):
- """Type Ctrl-D to exit the shell"""
- self.stdout.write('\n')
- # Exit the interpreter on ^D
- return True
-
- def onecmd(self, line):
- try:
- return cmd.Cmd.onecmd(self, line)
- except Exception, e:
- data = '\n'.join('% ' + l for l in str(e).splitlines())
- self.stdout.write(data + '\n\n')
-
- if self.debug:
- import traceback
- self.stdout.write(traceback.format_exc())
-
- def cmdloop(self, *args, **kwargs):
- while True:
- try:
- cmd.Cmd.cmdloop(self, *args, **kwargs)
- except KeyboardInterrupt:
- self.stdout.write('\n')
- # Abort the current command and continue on ^C
- continue
- break
-
- def runscript(self, script):
- """Run the specified file as a script."""
- for line in script:
- if not line.strip().startswith('#'):
- self.onecmd(line)
-
- def help_help(self):
- # You've got to be kidding me
- self.stdout.write('Type "help <topic>" for help on
commands\n')
-
- def __getattr__(self, name):
- """
- Get the do_ or complete_ method for a command from the current mode.
- The help_ methods should not be needed, as commands can provide help
- via their docstrings.
- """
-
- action, u, command_name = name.partition('_')
- if command_name:
- try:
- command = self.mode[command_name]
- except KeyError:
- pass
- else:
- if action == 'do':
- return command
- if action == 'complete':
- return command.complete
-
- raise AttributeError("%r object has no attribute %r" %
- (type(self).__name__, name))
-
- def get_names(self):
- return (filter(lambda n: n != 'do_EOF', cmd.Cmd.get_names(self)) +
- ['%s_%s' % (p, c) for p in ['do', 'complete',
'help']
- for c in self.mode])
-
-
-if __name__ == '__main__':
- mode = mode.Mode()
- shell = Interpreter('mhsh', mode)
- shell.cmdloop()
-
diff --git a/src/python/matahari/interpreter/mode.py
b/src/python/matahari/interpreter/mode.py
deleted file mode 100644
index a70dc50..0000000
--- a/src/python/matahari/interpreter/mode.py
+++ /dev/null
@@ -1,80 +0,0 @@
-"""
-Module for defining interpreter shell modes.
-
-To define a new mode, create a subclass of the Mode class. Customise the
-shell prompt by overriding the prompt() method. Add commands to the mode by
-concatenating them with the += operator.
-"""
-
-import types
-import time
-import command
-
-
-class Mode(object):
- """
- A interpreter shell mode. Each mode may have a distinctive prompt and its
- own set of commands, and may in addition store whatever state it likes.
- """
-
- def __init__(self, *commands):
- """Initialise with an initial set of commands."""
- self.shell = None
- self.commands = {}
- for m in dir(type(self)):
- try:
- c = getattr(self, m)
- except AttributeError:
- pass
- else:
- if (isinstance(c, command.CommandHandler)):
- self += c
- for c in commands:
- self += c
-
- def activate(self, interpreter):
- """Method called by the interpreter when the mode becomes
active."""
- self.shell = interpreter
-
- def deactivate(self, interpreter):
- """Method called by the interpreter when the mode becomes
inactive."""
- if self.shell == interpreter:
- self.shell = None
-
- def prompt(self):
- """
- Return any additions to the command prompt that are to be appended in
- this mode. This method may be overridden by subclasses.
- """
- return ''
-
- def write(self, *lines):
- """Write one or more lines to the console."""
- for line in lines:
- self.shell.stdout.write(line + '\n')
-
- def __iadd__(self, cmd):
- """Add a command."""
- n = cmd.name
- c = self.commands.get(n)
- self.commands[n] = c + cmd
- return self
-
- def __getitem__(self, key):
- return self.commands[key]
-
- def __iter__(self):
- return iter(self.commands)
-
- def iterkeys(self):
- return iter(self)
-
- @command.Command('sleep', float)
- def sleep(self, kw_sleep, seconds):
- """Sleep for the specified number of seconds"""
- time.sleep(seconds)
-
- @command.Command('quit')
- def quit(self, kw_quit):
- """Exit the interpreter"""
- return True
diff --git a/src/python/matahari/shell.py b/src/python/matahari/shell.py
deleted file mode 100644
index 6635320..0000000
--- a/src/python/matahari/shell.py
+++ /dev/null
@@ -1,12 +0,0 @@
-import api
-import shellmode
-import interpreter
-
-class MatahariShell(object):
- def __init__(self, prompt='mhsh'):
- self.state = api.Matahari()
- initial_mode = shellmode.RootMode(self.state)
- self.interpreter = interpreter.Interpreter(prompt, initial_mode, debug=True)
-
- def __call__(self):
- self.interpreter.cmdloop()
diff --git a/src/python/matahari/shell/__init__.py
b/src/python/matahari/shell/__init__.py
new file mode 100644
index 0000000..7ea8d2e
--- /dev/null
+++ b/src/python/matahari/shell/__init__.py
@@ -0,0 +1,12 @@
+import matahari.api
+import mode
+import interpreter
+
+class MatahariShell(object):
+ def __init__(self, prompt='mhsh'):
+ self.state = matahari.api.Matahari()
+ initial_mode = mode.RootMode(self.state)
+ self.interpreter = interpreter.Interpreter(prompt, initial_mode, debug=True)
+
+ def __call__(self):
+ self.interpreter.cmdloop()
diff --git a/src/python/matahari/shell/interpreter/__init__.py
b/src/python/matahari/shell/interpreter/__init__.py
new file mode 100644
index 0000000..96af66f
--- /dev/null
+++ b/src/python/matahari/shell/interpreter/__init__.py
@@ -0,0 +1,3 @@
+from interpreter import Interpreter
+from mode import Mode
+from command import Command, InvalidArgumentException
diff --git a/src/python/matahari/shell/interpreter/command.py
b/src/python/matahari/shell/interpreter/command.py
new file mode 100644
index 0000000..f857196
--- /dev/null
+++ b/src/python/matahari/shell/interpreter/command.py
@@ -0,0 +1,976 @@
+"""
+Module for defining the syntax of CLI commands.
+
+To define a command, create a handler function and wrap it with the Command
+decorator, like so:
+
+ @Command('test', 'PARAM')
+ def test(kw_test, param):
+ '''
+ A test command
+
+ Prints the parameter passed in.
+ '''
+ assert kw_test == 'test'
+ print param
+
+The parameters to the Command constructor are the arguments to the command.
+Arguments that are not in all caps are fixed keywords.
+
+Arguments in all caps (e.g. 'PARAM') are variable parameters. Additionally,
+an argument that is a function will be treated as a type conversion and
+validation function for a variable parameter. For example, pass the builtin
+int class to convert the argument to an integer. The function should throw
+a ValueError exception if the argument cannot be converted. If the function
+has an attribute named "complete", this will be called to provide suggested
+values for tab-completion. The complete method is passed the current
+fragment that the user is trying to tab-complete, and should return a list
+of matching allowable values.
+
+A group of optional parameters can be passed as a tuple. For example:
+
+ @Command('test2', ('optional-keyword', 'PARAM'),
'MANDATORY')
+ def test2(kw_test, (kw_optkw, param), mandatory):
+ '''A test command with optional parameters'''
+ if kw_optkw is not None:
+ print 'Optional:', param
+ print 'Mandatory:', mandatory
+
+Note that if the optional parameters are not supplied, they will each have
+the value None when passed to the handler function. This is done (instead of
+passing an empty list) so that tuple unpacking can be used as shown in the
+example. If a validation function has an attribute named "default", this
+default value will be used instead of None. If the default attribute is
+callable, it will be called to calculate the default.
+
+A variable argument list can be implemented by passing a list as the final
+parameter, thus:
+
+ @Command('test3', ['PARAMS'])
+ def test3(kw_test, *params):
+ '''A test command with variable arguments'''
+ print 'Params:' params
+
+A least one argument must be supplied for a variable argument list. To allow
+a list with zero items, make it optional by wrapping it in a tuple. In the case
+of variable argument lists, the default return value is an empty list (rather
+than None as for other parameters).
+
+The docstring of the command handler function will be used to display help
+about the command to the user on request.
+"""
+
+
+import types
+from itertools import *
+
+
+class Command(object):
+ """
+ Decorator to define the argument list for a command handler.
+
+ Arguments may be keywords (lower-case strings), string parameters
+ (upper-case strings), or parameters of different types (functions that
+ validate and/or convert the parameters from strings).
+
+ A series of arguments can be made optional by enclosing them in a tuple,
+ or repeated by enclosing in a list.
+ """
+
+ def __init__(self, name, *args):
+ """Initialise with a list of command arguments."""
+ self.name = Keyword(name)
+ self.args = [Argument(a) for a in args]
+
+ def __call__(self, func):
+ """Decorate the command handler."""
+ return CommandHandler(func, self.name, self.args)
+
+
+class InvalidArgumentException(Exception):
+ """
+ An Exception raised when the value of an argument passed by the user is not
+ valid in the current context.
+ """
+ pass
+
+class InvalidCommandException(Exception):
+ """
+ An Exception raised when a command executed by the user is invalid.
+ """
+
+ def __init__(self, line, candidates=[], message=None):
+ self.line = line
+ self.candidates = candidates
+ if message is None:
+ message = 'Invalid Command: "%s"' % self.line
+
+ if self.candidates:
+ message += ('\nCandidates are:\n' +
+ '\n'.join(' ' + str(c) for c, e in
self.candidates))
+
+ super(InvalidCommandException, self).__init__(message)
+
+
+class CommandHandler(object):
+ """A handler for a command."""
+
+ def __init__(self, func, name, args):
+ """Initialise with a handler function and a list of
arguments."""
+ self.do = func
+ self.name = str(name)
+ self.args = [name] + args
+ self.arg_graph = ArgGraph(self.args)
+ self.__name__ = self.do.__name__
+ self.__doc__ = self.help()
+
+ def help(self):
+ """Return the help string for the command."""
+ lines = [' > %s\n' % (str(self),)]
+ doc = self._trimdoc(self.do.__doc__)
+ if doc:
+ lines.append(doc + '\n')
+ return '\n'.join(lines)
+
+ def complete(self, text, line, begidx, endidx):
+ """Return a list of tab-completion options for the
command."""
+ values = line.split()
+ if values[-1] != text:
+ values.append(text)
+
+ match = lambda (a, v): a and a.complete(v) or []
+
+ def completions(arglist):
+ argvals = takewhile(lambda (a, v): v is not None,
+ izip_longest(arglist, values, fillvalue=None))
+ for m in imap(match, argvals):
+ if not m:
+ return []
+ return m
+
+ candidates = chain.from_iterable(imap(completions, self.arg_graph))
+
+ return list(set(candidates))
+
+ def __get__(self, obj, objtype=None):
+ # Allow the handler to be bound to objects as a method
+ boundfunc = types.MethodType(self.do, obj, objtype)
+ return CommandHandler(boundfunc, self.args[0], self.args[1:])
+
+ def __call__(self, line):
+ """Do the command."""
+ vals = [self.name] + line.split()
+ args = tuple(Argument.collate(self.args, vals))
+
+ if len(args) < len(self.args):
+ miss = ' '.join(str(a) for a in self.args[len(args):])
+ raise InvalidCommandException(line,
+ message="Missing arguments: '%s'" % miss)
+ if vals:
+ extra = ' '.join(vals)
+ raise InvalidCommandException(line,
+ message="Excess arguments: '%s'" % extra)
+
+ return self.do(*args)
+
+ def __repr__(self):
+ return '@Command(%s)' % ', '.join(repr(a) for a in self.args)
+
+ def __str__(self):
+ return ' '.join(str(a) for a in self.args)
+
+ def __add__(self, other):
+ if other is None:
+ return self
+ return CommandGroupHandler(self.name, self, other)
+
+ def __radd__(self, other):
+ if other is None:
+ return self
+ return CommandGroupHandler(self.name, other, self)
+
+ @staticmethod
+ def _trimdoc(docstring):
+ """Trim whitespace from a docstring."""
+ if not docstring:
+ return ''
+ # Convert tabs to spaces (following the normal Python rules)
+ # and split into a list of lines:
+ lines = docstring.expandtabs().splitlines()
+ # Determine minimum indentation (first line doesn't count):
+ nonblank = lambda l: l and not l.isspace()
+ indents = [len(l) - len(l.lstrip()) for l in lines[1:] if nonblank(l)]
+ indent = indents and reduce(min, indents) or 0
+ # Remove indentation (first line is special):
+ trimmed = [lines[0].strip()] + [l[indent:].rstrip() for l in lines[1:]]
+ # Strip off trailing and leading blank lines:
+ while trimmed and not trimmed[-1]:
+ trimmed.pop()
+ while trimmed and not trimmed[0]:
+ trimmed.pop(0)
+ # Return a single string:
+ return '\n'.join(trimmed)
+
+
+class CommandGroupHandler(object):
+ """A handler for a group of commands with the same
name."""
+
+ def __init__(self, *args):
+ """
+ Initialise with an optional command name and a series of command
+ handlers.
+ """
+ if isinstance(args[0], basestring):
+ self.name = args[0]
+ args = args[1:]
+ else:
+ self.name = args[0].name
+ self.cmds = []
+ self.__doc__ = None
+ for c in args:
+ self += c
+ self.__doc__ = self.help()
+
+ def help(self):
+ """Return the help string for the command."""
+ return '\n\n'.join(c.help() for c in self.cmds)
+
+ def complete(self, text, line, begidx, endidx):
+ """Return a list of tab-completion options for the
command."""
+ completions = lambda c: c.complete(text, line, begidx, endidx)
+ return list(set().union(*map(completions, self.cmds)))
+
+ def _check_add(self, other):
+ if not isinstance(other, CommandHandler):
+ raise TypeError("%s is not a CommandHandler" % type(other))
+ if other.name != self.name:
+ raise ValueError('Command name "%s" does not match group
"%s"' %
+ (other.name, self.name))
+
+ def __add__(self, other):
+ """Create a new group including an additional command or
group."""
+ if other is None:
+ return self
+ if isinstance(other, CommandGroupHandler):
+ cmds = other.cmds
+ if other.name != self.name:
+ raise ValueError('Command names "%s" and "%s" do
not match' %
+ (self.name, other.name))
+ else:
+ self._check_add(other)
+ cmds = [other]
+ return CommandGroupHandler(self.name, *(self.cmds + cmds))
+
+ def __iadd__(self, other):
+ """Add a new command to the group."""
+ self._check_add(other)
+ self.cmds.append(other)
+ if self.__doc__ is not None:
+ self.__doc__ = self.help()
+ return self
+
+ def __repr__(self):
+ return 'CommandGroupHandler(%s)' % repr(self.cmds)
+
+ def __str__(self):
+ return '\n'.join(' ' + str(c) for c in self.cmds)
+
+ def __call__(self, line):
+ """Do the command."""
+ failures = []
+
+ for c in self.cmds:
+ try:
+ return c(line)
+ except InvalidCommandException, e:
+ failures.extend(e.candidates or [(c, None)])
+ except InvalidArgumentException, e:
+ failures.append((c, e))
+
+ line = ' '.join((self.name, line))
+ raise InvalidCommandException(line.strip(), candidates=failures)
+
+
+class ArgGraph(object):
+ """A representation of the allowable sequences of command arguments in
the
+ form of a directed acyclic graph."""
+
+ def __init__(self, args):
+ """Initialise with a list of arguments"""
+ self.edges = {}
+ roots, leaves = self._addargs(args)
+ self.first = roots[0]
+
+ def __getitem__(self, key):
+ if key is None:
+ return self.first
+ return self.edges[key]
+
+ def __iter__(self):
+ """
+ Return a generator for all of the allowed sequences of arguments.
+ """
+ return self._getpaths()
+
+ def _getpaths(self, visited=None):
+ """
+ Return a generator for all of the allowed sequences of agruments
+ beginning with the given subpath.
+ """
+ children = self[visited and visited[-1]]
+
+ isleaf = lambda n: n not in self.edges
+ pathto = lambda n: (visited or []) + [n]
+
+ # Paths that end with the start node's children
+ leaf_paths = imap(lambda leaf: pathto(leaf),
+ ifilter(isleaf, children))
+
+ # Treat each Argument as an iterator and chain their output to pick up
+ # internal members of e.g. RepeatedArguments
+ arg_chains = imap(chain.from_iterable, leaf_paths)
+
+ getpaths = lambda n: self._getpaths(pathto(n))
+ nonleaf_children = ifilterfalse(isleaf, children)
+
+ # Paths that are recursively obtained via non-leaf children
+ child_arg_chains = chain.from_iterable(imap(getpaths,
+ nonleaf_children))
+
+ return chain(arg_chains, child_arg_chains)
+
+ def _addargs(self, args):
+ """Add a list of arguments to the Graph"""
+ roots_found = False
+ roots = []
+ leaves = []
+
+ optional = lambda n: isinstance(n, OptionalArguments)
+
+ def handle_opt_args(opt_args):
+ r, l = self._addargs(opt_args)
+ self._addedges(leaves, *r)
+ if not roots_found:
+ roots.extend(r)
+ leaves.extend(l)
+
+ def handle_arg(arg):
+ self._addedges(leaves, arg)
+ if not roots_found:
+ roots.append(arg)
+ leaves[:] = [arg]
+
+ for a in args:
+ if optional(a):
+ handle_opt_args(a)
+ else:
+ handle_arg(a)
+ roots_found = True
+
+ return roots, leaves
+
+ def _addedges(self, srcs, *dests):
+ for s in srcs:
+ l = self.edges.get(s, [])
+ l.extend(dests)
+ self.edges[s] = l
+
+
+class Argument(object):
+ """The definition of an argument to a command"""
+
+ def __new__(cls, arg):
+ """Create an argument of the correct type from its
definition."""
+ if cls != Argument:
+ return super(Argument, cls).__new__(cls)
+
+ if isinstance(arg, tuple):
+ argclass = OptionalArguments
+ elif isinstance(arg, list):
+ argclass = RepeatedArguments
+ elif callable(arg):
+ argclass = Parameter
+ elif isinstance(arg, basestring):
+ if arg.upper() == arg:
+ argclass = Parameter
+ else:
+ argclass = Keyword
+ else:
+ return arg
+ return argclass(arg)
+
+ @staticmethod
+ def collate(args, values):
+ """
+ Match values to arguments and return an argument list suitable for
+ passing to the handler function.
+ """
+ arglist = chain.from_iterable(imap(lambda a: a.match(values), args))
+ return takewhile(lambda p: p is not None, arglist)
+
+ def default(self):
+ return None
+
+ def __iter__(self):
+ yield self
+
+
+class Parameter(Argument):
+ """A parameter passed to the command by the user"""
+ def __init__(self, param):
+ self.param = param
+
+ def __repr__(self):
+ if callable(self.param):
+ if hasattr(self.param, '__name__'):
+ return self.param.__name__
+ else:
+ return type(self.param).__name__
+ return str(self.param)
+
+ def __str__(self):
+ return repr(self).upper()
+
+ def __call__(self, arg):
+ if callable(self.param):
+ try:
+ return self.param(arg)
+ except ValueError, e:
+ raise InvalidArgumentException('Invalid argument value (%s)' %
e)
+ return arg
+
+ def default(self):
+ if hasattr(self.param, 'default'):
+ if callable(self.param.default):
+ return self.param.default()
+ else:
+ return self.param.default
+ return None
+
+ def complete(self, value):
+ if hasattr(self.param, 'complete') and callable(self.param.complete):
+ return self.param.complete(value)
+ try:
+ if value:
+ parsed = self(value)
+ return [value and (value + ' ')]
+ except InvalidArgumentException:
+ return []
+
+ def match(self, values):
+ if not values:
+ yield None
+ return
+ yield self(values.pop(0))
+
+
+class Keyword(Argument):
+ """A fixed keyword that must be present in the
command"""
+ def __init__(self, kw):
+ self.kw = kw
+
+ def __str__(self):
+ return self.kw
+
+ def __eq__(self, other):
+ return self.kw == other
+
+ def __ne__(self, other):
+ return self.kw != other
+
+ def complete(self, value):
+ if self.kw.startswith(value):
+ return [self.kw + ' ']
+ else:
+ return []
+
+ def match(self, values):
+ if not values:
+ yield None
+ return
+ val = values[0]
+ if self != val:
+ raise InvalidArgumentException("Invalid keyword '%s'" %
val)
+ yield values.pop(0)
+
+
+class ArgumentList(Argument):
+ """Abstract base class for lists of arguments"""
+
+ def __init__(self, args):
+ self.args = [Argument(a) for a in args]
+
+ def __iter__(self):
+ return iter(self.args)
+
+ def __repr__(self):
+ return ', '.join(repr(a) for a in self.args)
+
+ def __str__(self):
+ return ' '.join(str(a) for a in self.args)
+
+
+class OptionalArguments(ArgumentList):
+ """A list of arguments that may be omitted when the command is
called"""
+
+ def __repr__(self):
+ contents = ArgumentList.__repr__(self)
+ if len(self.args) == 1:
+ contents += ','
+ return '(%s)' % contents
+
+ def __str__(self):
+ return '(%s)' % ArgumentList.__str__(self)
+
+ def default(self):
+ return [a.default() for a in self.args]
+
+ def match(self, values):
+ args = self.default()
+ if values:
+ values_copy = list(values)
+ try:
+ args = list(self.collate(self.args, values_copy))
+ values[:] = values_copy
+ except InvalidArgumentException:
+ pass
+ yield args
+
+
+class RepeatedArguments(ArgumentList):
+ """A list of arguments that may be repeated"""
+
+ def __iter__(self):
+ return cycle(self.args)
+
+ def __repr__(self):
+ return '[%s]' % ArgumentList.__repr__(self)
+
+ def __str__(self):
+ return '[%s]' % ArgumentList.__str__(self)
+
+ def match(self, values):
+ if not values:
+ raise InvalidArgumentException("Empty variadic argument list")
+
+ values_remain = lambda c: values
+
+ iterations = takewhile(values_remain,
+ starmap(self.collate,
+ repeat((self.args, values))))
+ return chain.from_iterable(iterations)
+
+ def default(self):
+ return []
+
+
+import unittest
+
+class CommandTest(unittest.TestCase):
+ class CallCounter(object):
+ def __init__(self, func):
+ self.func = func
+ self.__name__ = func.__name__
+ self.calls = 0
+
+ def __call__(self, *args, **kwargs):
+ self.calls += 1
+ return self.func(*args, **kwargs)
+
+
+ def setUp(self):
+ def nullCmd(*args, **kwargs):
+ self.fail('Command handler erroneously called')
+ self.nullCmd = nullCmd
+
+ def test_simple(self):
+ @self.CallCounter
+ def foo(arg):
+ self.assertEqual(arg, 'foo')
+ handler = Command('foo')(foo)
+ handler('')
+ self.assertEqual(foo.calls, 1)
+
+ def test_simple_bad(self):
+ handler = Command('foo')(self.nullCmd)
+ self.assertRaises(InvalidCommandException, handler, 'bar')
+
+ def test_param(self):
+ @self.CallCounter
+ def foo(kwfoo, param):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ handler = Command('foo', 'PARAM')(foo)
+ handler('bar')
+ self.assertEqual(foo.calls, 1)
+
+ def test_param_missing(self):
+ handler = Command('foo', 'PARAM')(self.nullCmd)
+ self.assertRaises(InvalidCommandException, handler, '')
+
+ def test_param_extra(self):
+ handler = Command('foo')(self.nullCmd)
+ self.assertRaises(InvalidCommandException, handler, 'bar')
+
+ def test_param_validator(self):
+ @self.CallCounter
+ def foo(kwfoo, intarg):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertIsInstance(intarg, int)
+ self.assertEqual(intarg, 42)
+ handler = Command('foo', int)(foo)
+ handler('42')
+ self.assertEqual(foo.calls, 1)
+
+ def test_param_validator_invalid(self):
+ @self.CallCounter
+ def foo(kwfoo, intarg):
+ self.fail('Command handler called')
+ handler = Command('foo', int)(foo)
+ self.assertRaises(InvalidArgumentException, handler, 'bar')
+
+ def test_opt_args(self):
+ @self.CallCounter
+ def foo(kwfoo, (kwbar, param1), param2):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(kwbar, 'bar')
+ self.assertEqual(param1, 'baz')
+ self.assertEqual(param2, 'blarg')
+ handler = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(foo)
+ handler('bar baz blarg')
+ self.assertEqual(foo.calls, 1)
+
+ def test_opt_args_missing(self):
+ @self.CallCounter
+ def foo(kwfoo, optargs, param2):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(optargs, [None, None])
+ self.assertEqual(param2, 'wibble')
+ handler = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(foo)
+ handler('wibble')
+ self.assertEqual(foo.calls, 1)
+
+ def test_opt_args_partial(self):
+ handler = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(self.nullCmd)
+ self.assertRaises(InvalidCommandException, handler, 'bar')
+
+ def test_opt_args_end_missing(self):
+ @self.CallCounter
+ def foo(kwfoo, optargs):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertSequenceEqual(optargs, [None])
+ handler = Command('foo', ('bar',))(foo)
+ handler('')
+ self.assertEqual(foo.calls, 1)
+
+ def test_opt_kw(self):
+ @self.CallCounter
+ def foo(kwfoo, (kwbar,), param):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(kwbar, 'bar')
+ self.assertEqual(param, 'baz')
+ handler = Command('foo', ('bar',), 'PARAM')(foo)
+ handler('bar baz')
+ self.assertEqual(foo.calls, 1)
+
+ def test_opt_kw_missing(self):
+ @self.CallCounter
+ def foo(kwfoo, (kwbar,), param):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertIs(kwbar, None)
+ self.assertEqual(param, 'quux')
+ handler = Command('foo', ('bar',), 'PARAM')(foo)
+ handler('quux')
+ self.assertEqual(foo.calls, 1)
+
+ def test_opt_kw_partial(self):
+ handler = Command('foo', ('bar',),
'PARAM')(self.nullCmd)
+ self.assertRaises(InvalidCommandException, handler, 'bar')
+
+ def test_opt_kw_missing(self):
+ @self.CallCounter
+ def foo(kwfoo, (kwbar,), param):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertIs(kwbar, None)
+ self.assertEqual(param, 'quux')
+ handler = Command('foo', ('bar',), 'PARAM')(foo)
+ handler('quux')
+ self.assertEqual(foo.calls, 1)
+
+ def test_list(self):
+ @self.CallCounter
+ def foo(kwfoo, param, *args):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ self.assertSequenceEqual(args, ['baz', 'blarg',
'wibble'])
+ handler = Command('foo', 'bar', ['PARAMS'])(foo)
+ handler('bar baz blarg wibble')
+ self.assertEqual(foo.calls, 1)
+
+ def test_list_single(self):
+ @self.CallCounter
+ def foo(kwfoo, param, *args):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ self.assertSequenceEqual(args, ['baz'])
+ handler = Command('foo', 'bar', ['PARAMS'])(foo)
+ handler('bar baz')
+ self.assertEqual(foo.calls, 1)
+
+ def test_list_empty(self):
+ handler = Command('foo', 'bar',
['PARAMS'])(self.nullCmd)
+ self.assertRaises(InvalidArgumentException, handler, 'bar')
+
+ def test_list_optional(self):
+ @self.CallCounter
+ def foo(kwfoo, param, args):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ self.assertSequenceEqual(args, [42, 43])
+ handler = Command('foo', 'bar', ([int],))(foo)
+ handler('bar 42 43')
+ self.assertEqual(foo.calls, 1)
+
+ def test_list_optional_empty(self):
+ @self.CallCounter
+ def foo(kwfoo, param, args):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ self.assertEqual(args[0], [])
+ handler = Command('foo', 'bar', ([int],))(foo)
+ handler('bar')
+ self.assertEqual(foo.calls, 1)
+
+ def test_list_optional_kw_err(self):
+ handler = Command('foo', ('bar',
['PARAMS']))(self.nullCmd)
+ self.assertRaises(InvalidCommandException, handler, 'bar')
+
+ def test_default_param(self):
+ class Answer(int):
+ default = 42
+
+ @self.CallCounter
+ def foo(kwfoo, param, args):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ self.assertEqual(args[0], 42)
+ handler = Command('foo', 'bar', (Answer,))(foo)
+ handler('bar')
+ self.assertEqual(foo.calls, 1)
+
+ def test_default_param_func(self):
+ class Answer(int):
+ @staticmethod
+ def default():
+ return 42
+
+ @self.CallCounter
+ def foo(kwfoo, param, args):
+ self.assertEqual(kwfoo, 'foo')
+ self.assertEqual(param, 'bar')
+ self.assertEqual(args[0], 42)
+ handler = Command('foo', 'bar', (Answer,))(foo)
+ handler('bar')
+ self.assertEqual(foo.calls, 1)
+
+
+class CompletionTest(unittest.TestCase):
+ def handler(self, *args):
+ @Command(*args)
+ def nullCmd(*args, **kwargs):
+ self.fail('Command handler called')
+ return nullCmd
+
+ def assertCompletion(self, handler, line, expected):
+ actual = self.complete(handler, line)
+ err = '\n' + '\n'.join(('Command: "%s"' %
handler,
+ 'Typed: "%s"' % line,
+ 'Completions: %s' % actual,
+ 'Expected: %s' % expected))
+ self.assertItemsEqual(actual, expected, err)
+
+ def complete(self, handler, line):
+ parts = line.rpartition(' ')
+ text = parts[2]
+ begidx = len(parts[0]) + len(parts[1])
+ endidx = len(line)
+ return handler.complete(text, line, begidx, endidx)
+
+ def test_kw(self):
+ h = self.handler('foo', 'bar')
+ self.assertCompletion(h, 'foo ', ['bar '])
+ self.assertCompletion(h, 'foo b', ['bar '])
+ self.assertCompletion(h, 'foo bar ', [])
+
+ def test_kw_space(self):
+ h = self.handler('foo', 'bar')
+ self.assertCompletion(h, 'foo', ['foo '])
+ self.assertCompletion(h, 'foo bar', ['bar '])
+
+ def test_multiple_kw(self):
+ h = self.handler('foo', 'bar', 'baz')
+ self.assertCompletion(h, 'foo b', ['bar '])
+ self.assertCompletion(h, 'foo bar ', ['baz '])
+
+ def test_param(self):
+ h = self.handler('foo', 'PARAM')
+ self.assertCompletion(h, 'foo ', [''])
+ self.assertCompletion(h, 'foo b', ['b '])
+ self.assertCompletion(h, 'foo bar ', [])
+
+ def test_param_validator(self):
+ h = self.handler('foo', int)
+ self.assertCompletion(h, 'foo ', [''])
+ self.assertCompletion(h, 'foo b', [])
+ self.assertCompletion(h, 'foo 1', ['1 '])
+ self.assertCompletion(h, 'foo 123 ', [])
+
+ def test_param_suggestor(self):
+ candidates = ['bar ', 'baz ', 'blarg ']
+ class Param(object):
+ def __call__(self):
+ pass
+ def complete(self, arg):
+ return [a for a in candidates if a.startswith(arg)]
+
+ h = self.handler('foo', Param())
+ self.assertCompletion(h, 'foo ', candidates)
+ self.assertCompletion(h, 'foo b', candidates)
+ self.assertCompletion(h, 'foo ba', ['bar ', 'baz '])
+ self.assertCompletion(h, 'foo bar', ['bar '])
+ self.assertCompletion(h, 'foo bar ', [])
+ self.assertCompletion(h, 'foo quux', [])
+
+ def test_multiple_param(self):
+ h = self.handler('foo', 'PARAM', 'PARAM2')
+ self.assertCompletion(h, 'foo ', [''])
+ self.assertCompletion(h, 'foo b', ['b '])
+ self.assertCompletion(h, 'foo bar ', [''])
+ self.assertCompletion(h, 'foo bar baz', ['baz '])
+ self.assertCompletion(h, 'foo bar baz ', [])
+
+ def test_opt_args(self):
+ h = self.handler('foo', ('bar', 'PARAM1'),
'PARAM2')
+ self.assertCompletion(h, 'foo ', ['bar ', ''])
+ self.assertCompletion(h, 'foo b', ['bar ', 'b '])
+ self.assertCompletion(h, 'foo bar ', [''])
+ self.assertCompletion(h, 'foo quux', ['quux '])
+ self.assertCompletion(h, 'foo quux ', [])
+
+ def test_multiple_opt_args(self):
+ h = self.handler('foo', ('bar', 'PARAM1'),
('baz', 'PARAM2'))
+ self.assertCompletion(h, 'foo', ['foo '])
+ self.assertCompletion(h, 'foo ', ['bar ', 'baz '])
+ self.assertCompletion(h, 'foo b', ['bar ', 'baz '])
+ self.assertCompletion(h, 'foo bar ', [''])
+ self.assertCompletion(h, 'foo baz ', [''])
+ self.assertCompletion(h, 'foo baz quux ', [])
+
+ def test_opt_kw(self):
+ h = self.handler('foo', ('bar',), 'PARAM')
+ self.assertCompletion(h, 'foo ', ['bar ', ''])
+ self.assertCompletion(h, 'foo b', ['bar ', 'b '])
+ self.assertCompletion(h, 'foo bar ', [''])
+ self.assertCompletion(h, 'foo bar quux', ['quux '])
+ self.assertCompletion(h, 'foo bar quux ', [])
+ self.assertCompletion(h, 'foo quux', ['quux '])
+ self.assertCompletion(h, 'foo quux ', [])
+
+ def test_multiple_opt_kw_first(self):
+ h = self.handler('foo', ('bar',), ('baz',))
+ self.assertCompletion(h, 'foo', ['foo '])
+ self.assertCompletion(h, 'foo ', ['bar ', 'baz '])
+ self.assertCompletion(h, 'foo b', ['bar ', 'baz '])
+ self.assertCompletion(h, 'foo bar ', ['baz '])
+ self.assertCompletion(h, 'foo baz ', [])
+ self.assertCompletion(h, 'foo quux ', [])
+ self.assertCompletion(h, 'foo bar quux ', [])
+
+ def test_multiple_opt_kw(self):
+ h = self.handler('foo', 'wibble', ('bar',),
('baz',))
+ self.assertCompletion(h, 'foo', ['foo '])
+ self.assertCompletion(h, 'foo ', ['wibble '])
+ self.assertCompletion(h, 'foo wibble ', ['bar ', 'baz
'])
+ self.assertCompletion(h, 'foo wibble b', ['bar ', 'baz
'])
+ self.assertCompletion(h, 'foo wibble bar ', ['baz '])
+ self.assertCompletion(h, 'foo wibble baz ', [])
+ self.assertCompletion(h, 'foo wibble quux ', [])
+ self.assertCompletion(h, 'foo wibble bar quux ', [])
+
+ def test_list(self):
+ h = self.handler('foo', 'bar', ['PARAMS'])
+ self.assertCompletion(h, 'foo ', ['bar '])
+ self.assertCompletion(h, 'foo bar', ['bar '])
+ self.assertCompletion(h, 'foo bar ', [''])
+ self.assertCompletion(h, 'foo bar baz', ['baz '])
+ self.assertCompletion(h, 'foo bar baz ', [''])
+ self.assertCompletion(h, 'foo bar baz blarg', ['blarg '])
+ self.assertCompletion(h, 'foo bar baz blarg ', [''])
+ self.assertCompletion(h, 'foo bar baz blarg wibble', ['wibble
'])
+ self.assertCompletion(h, 'foo bar baz blarg wibble ', [''])
+
+ def test_list_optional(self):
+ h = self.handler('foo', 'bar', (['PARAMS'],))
+ self.assertCompletion(h, 'foo ', ['bar '])
+ self.assertCompletion(h, 'foo bar', ['bar '])
+ self.assertCompletion(h, 'foo bar ', [''])
+ self.assertCompletion(h, 'foo bar baz', ['baz '])
+ self.assertCompletion(h, 'foo bar baz ', [''])
+ self.assertCompletion(h, 'foo bar baz blarg', ['blarg '])
+ self.assertCompletion(h, 'foo bar baz blarg ', [''])
+ self.assertCompletion(h, 'foo bar baz blarg wibble', ['wibble
'])
+ self.assertCompletion(h, 'foo bar baz blarg wibble ', [''])
+
+ def test_list_kw_optional(self):
+ h = self.handler('foo', 'bar', ('baz',
['PARAMS'],))
+ self.assertCompletion(h, 'foo ', ['bar '])
+ self.assertCompletion(h, 'foo bar', ['bar '])
+ self.assertCompletion(h, 'foo bar ', ['baz '])
+ self.assertCompletion(h, 'foo bar baz', ['baz '])
+ self.assertCompletion(h, 'foo bar baz ', [''])
+ self.assertCompletion(h, 'foo bar baz blarg', ['blarg '])
+ self.assertCompletion(h, 'foo bar baz blarg ', [''])
+ self.assertCompletion(h, 'foo bar baz blarg wibble', ['wibble
'])
+ self.assertCompletion(h, 'foo bar baz blarg wibble ', [''])
+
+
+class SyntaxTest(unittest.TestCase):
+ def setUp(self):
+ def nullCmd(*args, **kwargs):
+ self.fail('Command handler called')
+ self.nullCmd = nullCmd
+
+ def test_simple(self):
+ foo = Command('foo')(self.nullCmd)
+ self.assertEqual(str(foo), 'foo')
+
+ def test_param(self):
+ foo = Command('foo', 'PARAM')(self.nullCmd)
+ self.assertEqual(str(foo), 'foo PARAM')
+
+ def test_param_validator(self):
+ foo = Command('foo', int)(self.nullCmd)
+ self.assertEqual(str(foo), 'foo INT')
+
+ def test_opt_args(self):
+ foo = Command('foo', ('bar', 'PARAM1'),
'PARAM2')(self.nullCmd)
+ self.assertEqual(str(foo), 'foo (bar PARAM1) PARAM2')
+
+ def test_opt_kw(self):
+ foo = Command('foo', ('bar',), 'PARAM')(self.nullCmd)
+ self.assertEqual(str(foo), 'foo (bar) PARAM')
+
+ def test_list(self):
+ foo = Command('foo', ['PARAMS'])(self.nullCmd)
+ self.assertEqual(str(foo), 'foo [PARAMS]')
+
+ def test_opt_list(self):
+ foo = Command('foo', (['PARAMS'],))(self.nullCmd)
+ self.assertEqual(str(foo), 'foo ([PARAMS])')
+
+def suite():
+ suite = unittest.TestSuite()
+ suite.addTest(CommandTest)
+ suite.addTest(CompletionTest)
+ suite.addTest(SyntaxTest)
+ return suite
diff --git a/src/python/matahari/shell/interpreter/interpreter.py
b/src/python/matahari/shell/interpreter/interpreter.py
new file mode 100644
index 0000000..9eb1dbe
--- /dev/null
+++ b/src/python/matahari/shell/interpreter/interpreter.py
@@ -0,0 +1,125 @@
+"""
+Module defining a modal command interpreter.
+
+The commands supported by the interpreter are supplied by the current mode,
+which can be changed by passing a new mode to the set_mode() method.
+"""
+
+import cmd
+
+import mode
+from command import InvalidCommandException, InvalidArgumentException
+
+
+class Interpreter(cmd.Cmd):
+ """A modal command interpreter."""
+
+ def __init__(self, prompt, mode=mode.Mode(), debug=False):
+ """Initialise the interpreter in the initial
mode."""
+ cmd.Cmd.__init__(self)
+ self.name = prompt
+ self.debug = debug
+ self.mode = None
+ self.set_mode(mode)
+ self.doc_header = 'Commands:'
+ self.misc_header = 'Help topics:'
+ self.undoc_header = ''
+
+ def set_mode(self, mode):
+ """Set the shell mode."""
+ if self.mode is not None:
+ self.mode.deactivate(self)
+ self.mode = mode
+ self.mode.activate(self)
+ custom_prompt = mode.prompt()
+ self.prompt = self.name + (custom_prompt and ' ') + custom_prompt +
'> '
+
+ def completenames(self, *args):
+ # Tab-complete should move to the next argument
+ return [s + ' ' for s in cmd.Cmd.completenames(self, *args)]
+
+ def complete_help(self, *args):
+ # Don't suggest multiple arguments
+ if len(args[1].split()) > (args[1][-1] == ' ' and 1 or 2):
+ return []
+ # Explicitly call superclass's completenames() method, to avoid adding
+ # spaces (and thus duplicate output when combined with the topic list)
+ commands = set(cmd.Cmd.completenames(self, *args))
+ topics = set(a[5:] for a in self.get_names()
+ if a.startswith('help_' + args[0]))
+ return list(commands | topics)
+
+ def emptyline(self):
+ # Don't do anything for an empty command
+ pass
+
+ def do_EOF(self, line):
+ """Type Ctrl-D to exit the shell"""
+ self.stdout.write('\n')
+ # Exit the interpreter on ^D
+ return True
+
+ def onecmd(self, line):
+ try:
+ return cmd.Cmd.onecmd(self, line)
+ except Exception, e:
+ data = '\n'.join('% ' + l for l in str(e).splitlines())
+ self.stdout.write(data + '\n\n')
+
+ if self.debug:
+ import traceback
+ self.stdout.write(traceback.format_exc())
+
+ def cmdloop(self, *args, **kwargs):
+ while True:
+ try:
+ cmd.Cmd.cmdloop(self, *args, **kwargs)
+ except KeyboardInterrupt:
+ self.stdout.write('\n')
+ # Abort the current command and continue on ^C
+ continue
+ break
+
+ def runscript(self, script):
+ """Run the specified file as a script."""
+ for line in script:
+ if not line.strip().startswith('#'):
+ self.onecmd(line)
+
+ def help_help(self):
+ # You've got to be kidding me
+ self.stdout.write('Type "help <topic>" for help on
commands\n')
+
+ def __getattr__(self, name):
+ """
+ Get the do_ or complete_ method for a command from the current mode.
+ The help_ methods should not be needed, as commands can provide help
+ via their docstrings.
+ """
+
+ action, u, command_name = name.partition('_')
+ if command_name:
+ try:
+ command = self.mode[command_name]
+ except KeyError:
+ pass
+ else:
+ if action == 'do':
+ return command
+ if action == 'complete':
+ return command.complete
+
+ raise AttributeError("%r object has no attribute %r" %
+ (type(self).__name__, name))
+
+ def get_names(self):
+ return (filter(lambda n: n != 'do_EOF', cmd.Cmd.get_names(self)) +
+ ['%s_%s' % (p, c) for p in ['do', 'complete',
'help']
+ for c in self.mode])
+
+
+if __name__ == '__main__':
+ mode = mode.Mode()
+ shell = Interpreter('mhsh', mode)
+ shell.cmdloop()
+
diff --git a/src/python/matahari/shell/interpreter/mode.py
b/src/python/matahari/shell/interpreter/mode.py
new file mode 100644
index 0000000..a70dc50
--- /dev/null
+++ b/src/python/matahari/shell/interpreter/mode.py
@@ -0,0 +1,80 @@
+"""
+Module for defining interpreter shell modes.
+
+To define a new mode, create a subclass of the Mode class. Customise the
+shell prompt by overriding the prompt() method. Add commands to the mode by
+concatenating them with the += operator.
+"""
+
+import types
+import time
+import command
+
+
+class Mode(object):
+ """
+ A interpreter shell mode. Each mode may have a distinctive prompt and its
+ own set of commands, and may in addition store whatever state it likes.
+ """
+
+ def __init__(self, *commands):
+ """Initialise with an initial set of commands."""
+ self.shell = None
+ self.commands = {}
+ for m in dir(type(self)):
+ try:
+ c = getattr(self, m)
+ except AttributeError:
+ pass
+ else:
+ if (isinstance(c, command.CommandHandler)):
+ self += c
+ for c in commands:
+ self += c
+
+ def activate(self, interpreter):
+ """Method called by the interpreter when the mode becomes
active."""
+ self.shell = interpreter
+
+ def deactivate(self, interpreter):
+ """Method called by the interpreter when the mode becomes
inactive."""
+ if self.shell == interpreter:
+ self.shell = None
+
+ def prompt(self):
+ """
+ Return any additions to the command prompt that are to be appended in
+ this mode. This method may be overridden by subclasses.
+ """
+ return ''
+
+ def write(self, *lines):
+ """Write one or more lines to the console."""
+ for line in lines:
+ self.shell.stdout.write(line + '\n')
+
+ def __iadd__(self, cmd):
+ """Add a command."""
+ n = cmd.name
+ c = self.commands.get(n)
+ self.commands[n] = c + cmd
+ return self
+
+ def __getitem__(self, key):
+ return self.commands[key]
+
+ def __iter__(self):
+ return iter(self.commands)
+
+ def iterkeys(self):
+ return iter(self)
+
+ @command.Command('sleep', float)
+ def sleep(self, kw_sleep, seconds):
+ """Sleep for the specified number of seconds"""
+ time.sleep(seconds)
+
+ @command.Command('quit')
+ def quit(self, kw_quit):
+ """Exit the interpreter"""
+ return True
diff --git a/src/python/matahari/shell/mode.py b/src/python/matahari/shell/mode.py
new file mode 100644
index 0000000..7d05b5f
--- /dev/null
+++ b/src/python/matahari/shell/mode.py
@@ -0,0 +1,220 @@
+"""
+Matahari Shell modes.
+"""
+
+from interpreter import *
+import types
+
+
+QMF_ACCESS_MODES = (READ_CREATE, READ_WRITE, READ_ONLY) = range(1, 4)
+
+
+class RootMode(Mode):
+ """The default, root, shell mode."""
+
+ def __init__(self, state):
+ Mode.__init__(self)
+ self.state = state
+
+ select_wrapper = Command('select', 'host',
+ [types.Host(self.state)])
+ self += select_wrapper(self.select)
+
+ pkg_state = types.SharedInfo('org.matahariproject')
+ class_wrapper = Command('class',
+ ('package',
+ types.Package(self.state, pkg_state)),
+ types.Class(self.state, pkg_state))
+ self += class_wrapper(self.package_class)
+
+ def writelist(self, objects):
+ self.write(*[str(o) for o in objects])
+
+ def getFilteredMode(self):
+ return FilteredMode(self.state)
+
+ def getClassMode(self):
+ return ClassMode(self.state)
+
+ @Command('show', 'hosts')
+ def show_hosts(self, *kws):
+ """Show the list of connected hosts."""
+ self.writelist(self.state.list_hosts())
+
+ @Command('show', 'agents')
+ def show_agents(self, *kws):
+ """Show the list of connected agents."""
+ self.writelist(self.state.agents())
+
+ def select(self, kw_select, kw_host, *arg_hosts):
+ """ Select a set of hosts to restrict commands
to."""
+ self.state.set_hosts(arg_hosts)
+ self.shell.set_mode(self.getFilteredMode())
+
+ def package_class(self, kw_class, (kw_package, package), klass):
+ """ Select an object class to operate on."""
+ self.state.set_class(klass, package)
+ self.shell.set_mode(self.getClassMode())
+
+
+class FilteredMode(RootMode):
+ """Shell mode filtered by host"""
+
+ def __init__(self, state):
+ RootMode.__init__(self, state)
+
+ def getUnfilteredMode(self):
+ return RootMode(self.state)
+
+ def getClassMode(self):
+ return FilteredClassMode(self.state)
+
+ @Command('clear', 'selection')
+ def clear_selection(self, kw_clear, kw_selection):
+ """Clear the current host selection."""
+ self.state.set_hosts(None)
+ self.shell.set_mode(self.getUnfilteredMode())
+
+ @Command('show', 'selection')
+ def show_selection(self, *kws):
+ """Show the current host selection."""
+ self.writelist(self.state.hosts)
+
+ def prompt(self):
+ return '(host)'
+
+
+class CallMethodHandler(object):
+ """CLI Handler for a method call"""
+
+ def __new__(cls, mode, method):
+ handler = super(CallMethodHandler, cls).__new__(cls, mode, method)
+ CallMethodHandler.__init__(handler, mode, method)
+ wrapper = Command('call', handler._method.name, *handler._args)
+ return wrapper(handler)
+
+ def __init__(self, mode, method):
+ self._mode = mode
+ self._method = method
+ self._args = [types.QMFProperty(a) for a in self._method.arguments
+ if 'I' in a.dir]
+ self.__name__ = method.name
+ self.__doc__ = self.help()
+
+ def help(self):
+ lines = []
+ if self._method.desc is not None:
+ lines.append(self._method.desc)
+ if self._args:
+ lines.append('Parameters:')
+ lines.extend(a.help() for a in self._args)
+ return '\n'.join(lines)
+
+ def __call__(self, kw_call, methodname, *args):
+ """Call a method on the selected objects"""
+ qmf_method = getattr(self._mode.state.objects, methodname)
+ self._mode.writelist(qmf_method(*args))
+
+class GetPropertyHandler(object):
+ """CLI handler for getting an object property or
statistic"""
+
+ def __new__(cls, mode, prop, is_stat=False):
+ handler = super(GetPropertyHandler, cls).__new__(cls, mode,
+ prop, is_stat)
+ GetPropertyHandler.__init__(handler, mode, prop, is_stat)
+ wrapper = Command('get', repr(prop))
+ return wrapper(handler)
+
+ def __init__(self, mode, prop, is_stat=False):
+ self._mode = mode
+ self._prop = prop
+ self._is_stat = is_stat
+ self.__name__ = 'get_property'
+ self.__doc__ = self.help()
+
+ def __call__(self, kw_get, propname):
+ self._mode.writelist(getattr(self._mode.state.objects, propname))
+
+ def help(self):
+ doc = "Get the value of the %s %s for the selected objects"
+ desc = self._prop.desc and '\n\n' + self._prop.desc or ''
+ return (doc % (self._prop.name,
+ self._is_stat and 'statistic' or 'property')) +
desc
+
+class SetPropertyHandler(object):
+ """CLI handler for setting an object property"""
+
+ def __new__(cls, mode, prop):
+ handler = super(SetPropertyHandler, cls).__new__(cls, mode, prop)
+ SetPropertyHandler.__init__(handler, mode, prop)
+ wrapper = Command('set', repr(prop), handler._arg)
+ return wrapper(handler)
+
+ def __init__(self, mode, prop):
+ self._mode = mode
+ self._prop = prop
+ self._arg = types.QMFProperty(prop)
+ self.__name__ = 'set_property'
+ self.__doc__ = self.help()
+
+ def __call__(self, kw_set, propname, value):
+ setattr(self._mode.state.objects, propname, value)
+
+ def help(self):
+ doc = "Set the value of the %s property for the selected objects"
+ return (doc % (str(self._arg),)) + '\nDetails:\n' + self._arg.help()
+
+
+class ClassMode(RootMode):
+ """Shell mode filtered by QMF class."""
+
+ def __init__(self, state):
+ RootMode.__init__(self, state)
+
+ for prop in self.state.objects._properties():
+ self += GetPropertyHandler(self, prop)
+
+ if int(prop.access) == READ_WRITE:
+ self += SetPropertyHandler(self, prop)
+
+ for stat in self.state.objects._statistics():
+ self += GetPropertyHandler(self, stat, True)
+
+ for method in self.state.objects._methods():
+ self += CallMethodHandler(self, method)
+
+ def getFilteredMode(self):
+ return FilteredClassMode(self.state)
+
+ def getUnclassifiedMode(self):
+ return RootMode(self.state)
+
+ @Command('clear', 'class')
+ def clear_class(self, kw_clear, kw_class):
+ self.state.set_class(None, None)
+ self.shell.set_mode(self.getUnclassifiedMode())
+
+ @Command('select', 'property', 'PROPERTY', 'VALUE')
+ def select_property(self, kw_select, kw_property, arg_property, arg_value):
+ pass
+
+ def prompt(self):
+ return '[%s]' % (self.state.selected_classname(),)
+
+
+class FilteredClassMode(FilteredMode, ClassMode):
+ """Shell mode filtered by host and QMF class."""
+
+ def __init__(self, state):
+ FilteredMode.__init__(self, state)
+ ClassMode.__init__(self, state)
+
+ def getUnfilteredMode(self):
+ return ClassMode(self.state)
+
+ def getUnclassifiedMode(self):
+ return FilteredMode(self.state)
+
+ def prompt(self):
+ return FilteredMode.prompt(self) + ClassMode.prompt(self)
+
diff --git a/src/python/matahari/shell/types.py b/src/python/matahari/shell/types.py
new file mode 100644
index 0000000..e2d6de1
--- /dev/null
+++ b/src/python/matahari/shell/types.py
@@ -0,0 +1,312 @@
+import uuid
+
+class Host(object):
+ def __init__(self, state):
+ self.manager = state.manager
+
+ def __call__(self, param):
+ for h in self.manager.hosts():
+ if str(h) == param:
+ return h
+ raise ValueError('Host "%s" not found' % param)
+
+ def complete(self, param):
+ return filter(lambda s: s.startswith(param),
+ map(str, self.manager.hosts()))
+
+class SharedInfo(object):
+ def __init__(self, default=None):
+ self.last = self.default = default
+
+ def set(self, last):
+ self.last = last
+
+ def setdefault(self):
+ self.last = self.default
+
+ def get(self):
+ last, self.last = self.last, self.default
+ return last
+
+class Package(object):
+ def __init__(self, state, info):
+ self.state = state
+ self.info = info
+ self.info.setdefault()
+
+ def __call__(self, param):
+ if param not in self.state.list_packages():
+ self.info.setdefault()
+ raise ValueError('Package "%s" not found' % param)
+ self.info.set(param)
+ return param
+
+ def default(self):
+ self.info.setdefault()
+ return self.info.default
+
+ def complete(self, param):
+ packages = self.state.list_packages()
+ if param in packages:
+ self.info.set(param)
+ else:
+ self.info.setdefault()
+ return [p + ' ' for p in packages if p.startswith(param)]
+
+class Class(object):
+ def __init__(self, state, pkginfo):
+ self.state = state
+ self.pkginfo = pkginfo
+
+ def __call__(self, param):
+ package = self.pkginfo.get()
+ if param not in self.state.list_classnames(package):
+ fq = package is not None and ('%s:%s' % (package, param)) or param
+ raise ValueError('Class "%s" not found' % fq)
+ return param
+
+ def complete(self, param):
+ classnames = self.state.list_classnames(self.pkginfo.get())
+ return [c + ' ' for c in classnames if c.startswith(param)]
+
+
+QMF_TYPES = (U8, U16, U32, U64,
+ _UNUSED,
+ SSTR, LSTR,
+ ABSTIME, DELTATIME,
+ REF,
+ BOOL,
+ FLOAT, DOUBLE,
+ UUID,
+ FTABLE,
+ S8, S16, S32, S64,
+ OBJECT, LIST, ARRAY) = range(1, 23)
+
+class QMFProperty(object):
+ """CLI verification class for a QMF Property
argument"""
+
+ def __new__(cls, arg):
+ """Create a CLI argument of the correct type from a
SchemaArg."""
+ if cls == QMFProperty:
+ return _qmfPropertyTypes.get(arg.type, String)(arg)
+ else:
+ return super(QMFProperty, cls).__new__(cls, arg)
+
+ def __init__(self, arg):
+ self.arg = arg
+ self.__name__ = str(self)
+
+ def __repr__(self):
+ return self.__class__.__name__.lower()
+
+ def __str__(self):
+ return self.arg.name
+
+ def help(self):
+ return ('%9s %-18s %s' % (repr(self),
+ str(self),
+ self.arg.desc or '')).rstrip()
+
+ def default(self):
+ return self.arg.default
+
+class String(QMFProperty):
+ """A QMF String property argument"""
+ def __repr__(self):
+ if self.arg.type not in _qmfPropertyTypes:
+ return str(self)
+ return self.arg.type == SSTR and 'sstr' or 'lstr'
+
+ def __call__(self, param):
+ maxlen = self.arg.maxlen or (self.arg.type == SSTR and 255) or 65535
+ if len(param) > maxlen:
+ raise ValueError('Parameter is too long')
+ return param
+
+class Bool(QMFProperty):
+ """A QMF Boolean property argument"""
+
+ TRUTHY = ('true', 't', '1', 'y', 'yes')
+ FALSEY = ('false', 'f', '0', 'n', 'no')
+
+ def __call__(self, param):
+ lc = param.lower()
+ if lc in self.TRUTHY:
+ return True
+ if lc in self.FALSEY:
+ return False
+
+ try:
+ value = eval(param, {}, {})
+ except (NameError, SyntaxError):
+ raise ValueError('"%s" is not a boolean' % (param,))
+
+ if not isinstance(value, (int, bool)):
+ raise ValueError('"%s" is not a boolean' % (param,))
+
+ return bool(value)
+
+ def complete(self, param):
+ if not param:
+ return map(lambda l: l[0].capitalize(), (self.TRUTHY, self.FALSEY))
+ lc = param.lower()
+ matches = []
+ for v in self.TRUTHY + self.FALSEY:
+ if v == lc:
+ return [param + ' ']
+ if v.startswith(lc):
+ return [param + v[len(param):] + ' ']
+ return []
+
+class Int(QMFProperty):
+ """A QMF Integer property argument"""
+
+ NAMES = {U8: 'u8', U16: 'u16', U32: 'u32', U64:
'u64',
+ S8: 's8', S16: 's16', S32: 's32', S64:
's64'}
+ MINIMUM = {U8: 0, U16: 0,
+ U32: 0, U64: 0,
+ S8: -(1 << 7), S16: -(1 << 15),
+ S32: -(1 << 31), S64: -(1 << 63)}
+ MAXIMUM = {U8: (1 << 8) - 1, U16: (1 << 16) - 1,
+ U32: (1 << 32) - 1, U64: (1 << 64) - 1,
+ S8: (1 << 7) - 1, S16: (1 << 15) - 1,
+ S32: (1 << 31) - 1, S64: (1 << 63) - 1}
+
+ def __str__(self):
+ if self.arg.min is not None or self.arg.max is not None:
+ return '<%d-%d>' % (self._min(), self._max())
+
+ return QMFProperty.__str__(self)
+
+ def __repr__(self):
+ try:
+ return self.NAMES[self.arg.type]
+ except KeyError:
+ return QMFProperty.__repr__(self)
+
+ def _min(self):
+ """Get the minimum allowed value"""
+ if self.arg.min is not None:
+ return self.arg.min
+ try:
+ return self.MINIMUM[self.arg.type]
+ except KeyError:
+ return -(1 << 31)
+
+ def _max(self):
+ """Get the maximum allowed value"""
+ if self.arg.max is not None:
+ return self.arg.max
+ try:
+ return self.MAXIMUM[self.arg.type]
+ except KeyError:
+ return (1 << 31) - 1
+
+ def __call__(self, param):
+ value = int(param)
+ if value < self._min():
+ raise ValueError('Value %d underflows minimum of range (%d)' %
+ (value, self._min()))
+ if value > self._max():
+ raise ValueError('Value %d overflows maximum of range (%d)' %
+ (value, self._max()))
+ return value
+
+class Float(QMFProperty):
+ """A QMF Floating Point property argument"""
+
+ def __repr__(self):
+ return self.arg.type == FLOAT and 'float' or 'double'
+
+ def __call__(self, param):
+ return float(param)
+
+class Uuid(QMFProperty):
+ """A QMF UUID property argument"""
+
+ LENGTH = 32
+
+ def __call__(self, param):
+ return uuid.UUID(param)
+
+ def complete(self, param):
+ raw = param.replace('-', '')
+ try:
+ val = int(param, 16)
+ except ValueError:
+ return []
+ if len(raw) in (8, 12, 16, 20):
+ return [param + '-']
+ if len(raw) == self.LENGTH:
+ return [param + ' ']
+ return ['']
+
+class List(QMFProperty):
+ """A QMF List property argument"""
+
+ def __call__(self, param):
+ try:
+ l = eval(param, {}, {})
+ except (NameError, SyntaxError):
+ raise ValueError('"%s" is not a valid list' % (param,))
+ if not isinstance(l, list):
+ raise ValueError('"%s" is not a list' % (param,))
+ return l
+
+ def complete(self, param):
+ if not param:
+ return ['[']
+ if not param.startswith('['):
+ return []
+ if param.endswith(']'):
+ try:
+ self(param)
+ except ValueError:
+ return []
+ return [param + ' ']
+ return ['']
+
+class Map(QMFProperty):
+ """A QMF Map property argument"""
+
+ def __call__(self, param):
+ try:
+ l = eval(param, {}, {})
+ except (NameError, SyntaxError):
+ raise ValueError('"%s" is not a valid map' % (param,))
+ if not isinstance(l, dict):
+ raise ValueError('"%s" is not a map' % (param,))
+ return dict((unicode(k), v) for k, v in l.items())
+
+ def complete(self, param):
+ if not param:
+ return ['{']
+ if not param.startswith('{'):
+ return []
+ if param.endswith('}'):
+ try:
+ self(param)
+ except ValueError:
+ return []
+ return [param + ' ']
+ return ['']
+
+_qmfPropertyTypes = {
+ U8: Int,
+ U16: Int,
+ U32: Int,
+ U64: Int,
+ SSTR: String,
+ LSTR: String,
+ BOOL: Bool,
+ FLOAT: Float,
+ DOUBLE: Float,
+ UUID: Uuid,
+ FTABLE: Map,
+ S8: Int,
+ S16: Int,
+ S32: Int,
+ S64: Int,
+ LIST: List,
+ ARRAY: List,
+}
diff --git a/src/python/matahari/shellmode.py b/src/python/matahari/shellmode.py
deleted file mode 100644
index 10741df..0000000
--- a/src/python/matahari/shellmode.py
+++ /dev/null
@@ -1,220 +0,0 @@
-"""
-Matahari Shell modes.
-"""
-
-from interpreter import *
-import shelltypes as types
-
-
-QMF_ACCESS_MODES = (READ_CREATE, READ_WRITE, READ_ONLY) = range(1, 4)
-
-
-class RootMode(Mode):
- """The default, root, shell mode."""
-
- def __init__(self, state):
- Mode.__init__(self)
- self.state = state
-
- select_wrapper = Command('select', 'host',
- [types.Host(self.state)])
- self += select_wrapper(self.select)
-
- pkg_state = types.SharedInfo('org.matahariproject')
- class_wrapper = Command('class',
- ('package',
- types.Package(self.state, pkg_state)),
- types.Class(self.state, pkg_state))
- self += class_wrapper(self.package_class)
-
- def writelist(self, objects):
- self.write(*[str(o) for o in objects])
-
- def getFilteredMode(self):
- return FilteredMode(self.state)
-
- def getClassMode(self):
- return ClassMode(self.state)
-
- @Command('show', 'hosts')
- def show_hosts(self, *kws):
- """Show the list of connected hosts."""
- self.writelist(self.state.list_hosts())
-
- @Command('show', 'agents')
- def show_agents(self, *kws):
- """Show the list of connected agents."""
- self.writelist(self.state.agents())
-
- def select(self, kw_select, kw_host, *arg_hosts):
- """ Select a set of hosts to restrict commands
to."""
- self.state.set_hosts(arg_hosts)
- self.shell.set_mode(self.getFilteredMode())
-
- def package_class(self, kw_class, (kw_package, package), klass):
- """ Select an object class to operate on."""
- self.state.set_class(klass, package)
- self.shell.set_mode(self.getClassMode())
-
-
-class FilteredMode(RootMode):
- """Shell mode filtered by host"""
-
- def __init__(self, state):
- RootMode.__init__(self, state)
-
- def getUnfilteredMode(self):
- return RootMode(self.state)
-
- def getClassMode(self):
- return FilteredClassMode(self.state)
-
- @Command('clear', 'selection')
- def clear_selection(self, kw_clear, kw_selection):
- """Clear the current host selection."""
- self.state.set_hosts(None)
- self.shell.set_mode(self.getUnfilteredMode())
-
- @Command('show', 'selection')
- def show_selection(self, *kws):
- """Show the current host selection."""
- self.writelist(self.state.hosts)
-
- def prompt(self):
- return '(host)'
-
-
-class CallMethodHandler(object):
- """CLI Handler for a method call"""
-
- def __new__(cls, mode, method):
- handler = super(CallMethodHandler, cls).__new__(cls, mode, method)
- CallMethodHandler.__init__(handler, mode, method)
- wrapper = Command('call', handler._method.name, *handler._args)
- return wrapper(handler)
-
- def __init__(self, mode, method):
- self._mode = mode
- self._method = method
- self._args = [types.QMFProperty(a) for a in self._method.arguments
- if 'I' in a.dir]
- self.__name__ = method.name
- self.__doc__ = self.help()
-
- def help(self):
- lines = []
- if self._method.desc is not None:
- lines.append(self._method.desc)
- if self._args:
- lines.append('Parameters:')
- lines.extend(a.help() for a in self._args)
- return '\n'.join(lines)
-
- def __call__(self, kw_call, methodname, *args):
- """Call a method on the selected objects"""
- qmf_method = getattr(self._mode.state.objects, methodname)
- self._mode.writelist(qmf_method(*args))
-
-class GetPropertyHandler(object):
- """CLI handler for getting an object property or
statistic"""
-
- def __new__(cls, mode, prop, is_stat=False):
- handler = super(GetPropertyHandler, cls).__new__(cls, mode,
- prop, is_stat)
- GetPropertyHandler.__init__(handler, mode, prop, is_stat)
- wrapper = Command('get', repr(prop))
- return wrapper(handler)
-
- def __init__(self, mode, prop, is_stat=False):
- self._mode = mode
- self._prop = prop
- self._is_stat = is_stat
- self.__name__ = 'get_property'
- self.__doc__ = self.help()
-
- def __call__(self, kw_get, propname):
- self._mode.writelist(getattr(self._mode.state.objects, propname))
-
- def help(self):
- doc = "Get the value of the %s %s for the selected objects"
- desc = self._prop.desc and '\n\n' + self._prop.desc or ''
- return (doc % (self._prop.name,
- self._is_stat and 'statistic' or 'property')) +
desc
-
-class SetPropertyHandler(object):
- """CLI handler for setting an object property"""
-
- def __new__(cls, mode, prop):
- handler = super(SetPropertyHandler, cls).__new__(cls, mode, prop)
- SetPropertyHandler.__init__(handler, mode, prop)
- wrapper = Command('set', repr(prop), handler._arg)
- return wrapper(handler)
-
- def __init__(self, mode, prop):
- self._mode = mode
- self._prop = prop
- self._arg = types.QMFProperty(prop)
- self.__name__ = 'set_property'
- self.__doc__ = self.help()
-
- def __call__(self, kw_set, propname, value):
- setattr(self._mode.state.objects, propname, value)
-
- def help(self):
- doc = "Set the value of the %s property for the selected objects"
- return (doc % (str(self._arg),)) + '\nDetails:\n' + self._arg.help()
-
-
-class ClassMode(RootMode):
- """Shell mode filtered by QMF class."""
-
- def __init__(self, state):
- RootMode.__init__(self, state)
-
- for prop in self.state.objects._properties():
- self += GetPropertyHandler(self, prop)
-
- if int(prop.access) == READ_WRITE:
- self += SetPropertyHandler(self, prop)
-
- for stat in self.state.objects._statistics():
- self += GetPropertyHandler(self, stat, True)
-
- for method in self.state.objects._methods():
- self += CallMethodHandler(self, method)
-
- def getFilteredMode(self):
- return FilteredClassMode(self.state)
-
- def getUnclassifiedMode(self):
- return RootMode(self.state)
-
- @Command('clear', 'class')
- def clear_class(self, kw_clear, kw_class):
- self.state.set_class(None, None)
- self.shell.set_mode(self.getUnclassifiedMode())
-
- @Command('select', 'property', 'PROPERTY', 'VALUE')
- def select_property(self, kw_select, kw_property, arg_property, arg_value):
- pass
-
- def prompt(self):
- return '[%s]' % (self.state.selected_classname(),)
-
-
-class FilteredClassMode(FilteredMode, ClassMode):
- """Shell mode filtered by host and QMF class."""
-
- def __init__(self, state):
- FilteredMode.__init__(self, state)
- ClassMode.__init__(self, state)
-
- def getUnfilteredMode(self):
- return ClassMode(self.state)
-
- def getUnclassifiedMode(self):
- return FilteredMode(self.state)
-
- def prompt(self):
- return FilteredMode.prompt(self) + ClassMode.prompt(self)
-
diff --git a/src/python/matahari/shelltypes.py b/src/python/matahari/shelltypes.py
deleted file mode 100644
index e2d6de1..0000000
--- a/src/python/matahari/shelltypes.py
+++ /dev/null
@@ -1,312 +0,0 @@
-import uuid
-
-class Host(object):
- def __init__(self, state):
- self.manager = state.manager
-
- def __call__(self, param):
- for h in self.manager.hosts():
- if str(h) == param:
- return h
- raise ValueError('Host "%s" not found' % param)
-
- def complete(self, param):
- return filter(lambda s: s.startswith(param),
- map(str, self.manager.hosts()))
-
-class SharedInfo(object):
- def __init__(self, default=None):
- self.last = self.default = default
-
- def set(self, last):
- self.last = last
-
- def setdefault(self):
- self.last = self.default
-
- def get(self):
- last, self.last = self.last, self.default
- return last
-
-class Package(object):
- def __init__(self, state, info):
- self.state = state
- self.info = info
- self.info.setdefault()
-
- def __call__(self, param):
- if param not in self.state.list_packages():
- self.info.setdefault()
- raise ValueError('Package "%s" not found' % param)
- self.info.set(param)
- return param
-
- def default(self):
- self.info.setdefault()
- return self.info.default
-
- def complete(self, param):
- packages = self.state.list_packages()
- if param in packages:
- self.info.set(param)
- else:
- self.info.setdefault()
- return [p + ' ' for p in packages if p.startswith(param)]
-
-class Class(object):
- def __init__(self, state, pkginfo):
- self.state = state
- self.pkginfo = pkginfo
-
- def __call__(self, param):
- package = self.pkginfo.get()
- if param not in self.state.list_classnames(package):
- fq = package is not None and ('%s:%s' % (package, param)) or param
- raise ValueError('Class "%s" not found' % fq)
- return param
-
- def complete(self, param):
- classnames = self.state.list_classnames(self.pkginfo.get())
- return [c + ' ' for c in classnames if c.startswith(param)]
-
-
-QMF_TYPES = (U8, U16, U32, U64,
- _UNUSED,
- SSTR, LSTR,
- ABSTIME, DELTATIME,
- REF,
- BOOL,
- FLOAT, DOUBLE,
- UUID,
- FTABLE,
- S8, S16, S32, S64,
- OBJECT, LIST, ARRAY) = range(1, 23)
-
-class QMFProperty(object):
- """CLI verification class for a QMF Property
argument"""
-
- def __new__(cls, arg):
- """Create a CLI argument of the correct type from a
SchemaArg."""
- if cls == QMFProperty:
- return _qmfPropertyTypes.get(arg.type, String)(arg)
- else:
- return super(QMFProperty, cls).__new__(cls, arg)
-
- def __init__(self, arg):
- self.arg = arg
- self.__name__ = str(self)
-
- def __repr__(self):
- return self.__class__.__name__.lower()
-
- def __str__(self):
- return self.arg.name
-
- def help(self):
- return ('%9s %-18s %s' % (repr(self),
- str(self),
- self.arg.desc or '')).rstrip()
-
- def default(self):
- return self.arg.default
-
-class String(QMFProperty):
- """A QMF String property argument"""
- def __repr__(self):
- if self.arg.type not in _qmfPropertyTypes:
- return str(self)
- return self.arg.type == SSTR and 'sstr' or 'lstr'
-
- def __call__(self, param):
- maxlen = self.arg.maxlen or (self.arg.type == SSTR and 255) or 65535
- if len(param) > maxlen:
- raise ValueError('Parameter is too long')
- return param
-
-class Bool(QMFProperty):
- """A QMF Boolean property argument"""
-
- TRUTHY = ('true', 't', '1', 'y', 'yes')
- FALSEY = ('false', 'f', '0', 'n', 'no')
-
- def __call__(self, param):
- lc = param.lower()
- if lc in self.TRUTHY:
- return True
- if lc in self.FALSEY:
- return False
-
- try:
- value = eval(param, {}, {})
- except (NameError, SyntaxError):
- raise ValueError('"%s" is not a boolean' % (param,))
-
- if not isinstance(value, (int, bool)):
- raise ValueError('"%s" is not a boolean' % (param,))
-
- return bool(value)
-
- def complete(self, param):
- if not param:
- return map(lambda l: l[0].capitalize(), (self.TRUTHY, self.FALSEY))
- lc = param.lower()
- matches = []
- for v in self.TRUTHY + self.FALSEY:
- if v == lc:
- return [param + ' ']
- if v.startswith(lc):
- return [param + v[len(param):] + ' ']
- return []
-
-class Int(QMFProperty):
- """A QMF Integer property argument"""
-
- NAMES = {U8: 'u8', U16: 'u16', U32: 'u32', U64:
'u64',
- S8: 's8', S16: 's16', S32: 's32', S64:
's64'}
- MINIMUM = {U8: 0, U16: 0,
- U32: 0, U64: 0,
- S8: -(1 << 7), S16: -(1 << 15),
- S32: -(1 << 31), S64: -(1 << 63)}
- MAXIMUM = {U8: (1 << 8) - 1, U16: (1 << 16) - 1,
- U32: (1 << 32) - 1, U64: (1 << 64) - 1,
- S8: (1 << 7) - 1, S16: (1 << 15) - 1,
- S32: (1 << 31) - 1, S64: (1 << 63) - 1}
-
- def __str__(self):
- if self.arg.min is not None or self.arg.max is not None:
- return '<%d-%d>' % (self._min(), self._max())
-
- return QMFProperty.__str__(self)
-
- def __repr__(self):
- try:
- return self.NAMES[self.arg.type]
- except KeyError:
- return QMFProperty.__repr__(self)
-
- def _min(self):
- """Get the minimum allowed value"""
- if self.arg.min is not None:
- return self.arg.min
- try:
- return self.MINIMUM[self.arg.type]
- except KeyError:
- return -(1 << 31)
-
- def _max(self):
- """Get the maximum allowed value"""
- if self.arg.max is not None:
- return self.arg.max
- try:
- return self.MAXIMUM[self.arg.type]
- except KeyError:
- return (1 << 31) - 1
-
- def __call__(self, param):
- value = int(param)
- if value < self._min():
- raise ValueError('Value %d underflows minimum of range (%d)' %
- (value, self._min()))
- if value > self._max():
- raise ValueError('Value %d overflows maximum of range (%d)' %
- (value, self._max()))
- return value
-
-class Float(QMFProperty):
- """A QMF Floating Point property argument"""
-
- def __repr__(self):
- return self.arg.type == FLOAT and 'float' or 'double'
-
- def __call__(self, param):
- return float(param)
-
-class Uuid(QMFProperty):
- """A QMF UUID property argument"""
-
- LENGTH = 32
-
- def __call__(self, param):
- return uuid.UUID(param)
-
- def complete(self, param):
- raw = param.replace('-', '')
- try:
- val = int(param, 16)
- except ValueError:
- return []
- if len(raw) in (8, 12, 16, 20):
- return [param + '-']
- if len(raw) == self.LENGTH:
- return [param + ' ']
- return ['']
-
-class List(QMFProperty):
- """A QMF List property argument"""
-
- def __call__(self, param):
- try:
- l = eval(param, {}, {})
- except (NameError, SyntaxError):
- raise ValueError('"%s" is not a valid list' % (param,))
- if not isinstance(l, list):
- raise ValueError('"%s" is not a list' % (param,))
- return l
-
- def complete(self, param):
- if not param:
- return ['[']
- if not param.startswith('['):
- return []
- if param.endswith(']'):
- try:
- self(param)
- except ValueError:
- return []
- return [param + ' ']
- return ['']
-
-class Map(QMFProperty):
- """A QMF Map property argument"""
-
- def __call__(self, param):
- try:
- l = eval(param, {}, {})
- except (NameError, SyntaxError):
- raise ValueError('"%s" is not a valid map' % (param,))
- if not isinstance(l, dict):
- raise ValueError('"%s" is not a map' % (param,))
- return dict((unicode(k), v) for k, v in l.items())
-
- def complete(self, param):
- if not param:
- return ['{']
- if not param.startswith('{'):
- return []
- if param.endswith('}'):
- try:
- self(param)
- except ValueError:
- return []
- return [param + ' ']
- return ['']
-
-_qmfPropertyTypes = {
- U8: Int,
- U16: Int,
- U32: Int,
- U64: Int,
- SSTR: String,
- LSTR: String,
- BOOL: Bool,
- FLOAT: Float,
- DOUBLE: Float,
- UUID: Uuid,
- FTABLE: Map,
- S8: Int,
- S16: Int,
- S32: Int,
- S64: Int,
- LIST: List,
- ARRAY: List,
-}