yellow team mailing list archive
-
yellow team
-
Mailing list archive
-
Message #01069
[Merge] lp:~frankban/lpsetup/out-of-control into lp:lpsetup
Francesco Banconi has proposed merging lp:~frankban/lpsetup/out-of-control into lp:lpsetup.
Requested reviews:
Yellow Squad (yellow)
For more details, see:
https://code.launchpad.net/~frankban/lpsetup/out-of-control/+merge/118982
= Summary =
This branch is the result of a previous experimental branch. The goal is to reduce the inversion of control taking place in the argparser/subcommands machinery.
Before:
We had a customized *ArgumentParser* being able to register subcommands.
Subcommands were subclasses of *BaseSubCommand*, registered using
*ArgumentParser.register_subcommand*. From that point on, the subcommands
framework took care of the process of validating, restarting as root and
actual running each sub command.
In this branch:
The *ArgumentParser* is more or less a default *argparse.ArgumentParser*.
The only change there is a method override done to store actions in
a "public" attribute. Subcommands are just objects implementing a pre-defined
protocol, that is described in the branch. So subcommands can be instances,
or just modules.
The process of registering and running a subcommand is now handled by functions
in `cli.py`, and the code flow should be easier to follow because several actions
are now executed in a procedural way: prepare the parser, add common arguments,
add user-interaction related arguments, restart as root, namespace initialization
and validation, ... However, the inversion of control is still present, but it's
moved forward to the point of executing actual subcommand stuff: i.e. when
*subcommand.run* is called.
As a consequence, while the subcommand base classes are still there, they are now
more lightweight, and they represent only a flexible and easy way to implement
the subcommand protocol. They still introduce and handle the concepts of handlers
and steps, which are more well-documented in the code.
If we will decide in the future to follow the path of further reducing the inversion
of control (e.g. getting rid of subcommands base classes), this branch could be
considered as an incremental step.
For now, my opinion is that this could be a good trade off between the need for simplicity
and the generality/"DRYness" of OOP.
This branch adds documentation and massively reorganizes code: methods are now
functions, tests have been changed accordingly, some unused modules have been removed,
etc... so the diff is HUGE, and I am very sorry.
The integration test `test_install_lxc.py` passes.
== Other changes ==
Introduced a basic documentation for subcommands.
Modified *ArgumentParser* and base subcommand classes as described above. Also
removed no longer valid doctests. The code exercised by removed doctests is fully
covered by unittests.
*subcommand.get_description(namespace) is now called to display subcommands'
descriptions in a generic way: steps based subcommands override that method so that
the resulting description is created collecting steps' descriptions.
The `argparser` module now defines several functions explicitly called by cli:
*add_common_arguments*, *get_args_from_namespace*, *init_namespace*, *restart_as_root*.
The `cli `module itself now contains some helper functions:
- *get_parse*r returns an argument parser with all subcommands registered
- *prepare_parser* takes a parser (or a subparser) and a subcommand, adds common
arguments to the parsers and registers a callback pointing to the subcommand.
Note that the callback is a closure: that way we can immediately regain the
control of the code flow. Also note that this function can be used passing an
arbitrary parser. Usually a subparser is passed, but providing a normal
*ArgumentParser* this function could be reused to prepare a parser executing
separate subcommands (e.g. `lp-init-host` or `lp-init-repo`).
- *run*: taking care of:
- namespace initialization and validation
- interactive arguments handling
- executing a dry run, if requested
- restarting the same sub command as root, if required
- actual sub command execution
Removed the subcommand dynamic dispatcher: RIP.
This is how we usually define steps::
steps = [(step1, 'arg1', 'arg2'), (step2...)]
In this branch:
steps = [(step1, ['arg1', 'arg2']), (step2, []...)]
So the second item of each sequence is now a sequence of arg names, and a third
optional element is also allowed: a filter function that takes a namespace and must
return True if the step can be executed, False otherwise. As a side note, thanks
to this change we can now be more precise when collecting steps' descriptions.
Updated all the subcommands accordingly.
s/needs_root/root_required: needs_root is now a callable, part of the subcommands protocol.
Removed the `lpsetup.tests.examples` module: several steps and subcommands were defined there.
Now we have two factory test mixins in `lpsetup.tests.utils` that are used to make
steps or subcommands when needed in tests.
Added a new context manager (*call_replaced_by*) that temporarily mocks *subprocess.call*.
This is used to test the "restart as root" functions.
Removed the *root_not_needed* context manager: it was only used in dry run smoke tests:
now the dry execution does not want to restart as root, because it is handled before
`sudo` is called.
Removed no longer used *SubCommandTestMixin*, and fixed *StepsBasedSubCommandTestMixin*:
the latter can be safely removed once we get rid of the copy/paste smoke tests.
Moved *get_step_description* from `lpsetup.utils` to `lpsetup.argparser`.
Bumped version up a little.
--
https://code.launchpad.net/~frankban/lpsetup/out-of-control/+merge/118982
Your team Yellow Squad is requested to review the proposed merge of lp:~frankban/lpsetup/out-of-control into lp:lpsetup.
=== modified file 'lpsetup/__init__.py'
--- lpsetup/__init__.py 2012-06-28 10:33:46 +0000
+++ lpsetup/__init__.py 2012-08-09 15:36:20 +0000
@@ -9,7 +9,7 @@
'get_version',
]
-VERSION = (0, 2, 2)
+VERSION = (0, 2, 3)
def get_version():
=== modified file 'lpsetup/argparser.py'
--- lpsetup/argparser.py 2012-07-30 12:39:54 +0000
+++ lpsetup/argparser.py 2012-08-09 15:36:20 +0000
@@ -2,25 +2,158 @@
# Copyright 2012 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
-"""Argparse wrapper to add support for sub commands."""
+"""Helpers and functions to create sub commands.
+
+
+= Subcommands =
+
+A subcommand can be any object implementing the following protocol:
+
+ Required names:
+
+ - help: the command line help of the subcommand.
+
+ - get_description(namespace): a callable that receives a namespace
+ and must return the description of the subcommand as a string that
+ will be printed during interactive or dry execution.
+ Note that this name must only be defined if the sub command supports
+ interactive or dry run (see below in "Optional names").
+
+ - needs_root(namespace): a callable returning True if the sub command
+ requires the user to be root, False otherwise.
+
+ - run(namespace): a callable actually executing the sub command.
+
+ Optional names:
+
+ - add_arguments(parser): a callable that receives a parser and can
+ add sub command specific arguments to that parser.
+
+ - has_dry_run: True if the sub command supports dry execution,
+ False otherwise. If not defined, False is considered.
+
+ - has_interactive_run: True if the sub command supports interactive
+ execution, False otherwise. If not defined, the sub command is
+ considered non-interactive.
+
+ - prepare_namespace(namespace): a callable that receives a namespace
+ object. A sub command can define this callable to further initialize
+ or validate the namespace as needed. A ValidationError can be
+ raised by *prepare_namespace* to stop the sub command execution.
+
+
+However, this module provides two classes that can be used to automate the
+process in several ways.
+
+
+== BaseSubCommand ==
+
+This base class implements:
+
+ - help: as an empty string.
+
+ - prepare_namespace(namespace):
+
+ This base subcommand introduces the concept of handlers. Handlers are
+ callables receiving the current namespace. They can manipulate the
+ namespace adding or changing names. They can also validate the
+ namespace, e.g. when the validation depends on multiple args.
+ If something is wrong with the arguments, a handler should raise a
+ *ValidationError*.
+
+ Subcommands can store handlers in the *handlers* attribute as a
+ sequence, or can dynamically set up handlers overriding the
+ *get_handlers* method.
+
+ In any case, the *prepare_namespace* implementation of this class
+ calls the handlers in the order they are provided.
+
+ - needs_root(namespace):
+
+ returning *subcommand.root_required*, False by default.
+ This way subcommands can just set up the *root_required* attribute
+ if they don't need to calculate that value dynamically.
+
+ - get_description(namespace): returning *subcommand.help*
+
+All the required names in the subcommands protocol are implemented by
+*BaseSubCommand* but *run*, that must be overridden by subclasses to
+do actual stuff.
+
+
+== StepsBasedSubCommand ==
+
+Instances of this class can take advantage of the integrated "steps"
+machinery. The *run* method is implemented to execute steps in order.
+The main advantage of basing a subcommand on steps is the possibility
+of executing only a subset of the subcommand, including or skipping
+some steps. This allows more granularity when the subcommand is run.
+In particular, this base class implements:
+
+ - steps:
+
+ a sequence of *(callable, arg_names, filter-callable)* where:
+
+ - *arg_names* are the names of namespace attributes, used to retrieve
+ the corresponding values and generate a tuple of *args* to be
+ passed to the step
+ - callable is a callable step that will be run passing collected *args*
+ - *filter-callable* is an OPTIONAL function that takes the namespace
+ and returns True if the step must be run, False otherwise.
+ If not provided, the step is run.
+
+ Example: run *mystep*, passing *foo* and *bar* values,
+ only if the namespace contains 'foo' and 'foo' is True::
+
+ def my_step(foo, bar):
+ pass
+
+ steps = [
+ (my_step, ['foo', 'bar'],
+ lambda namespace: getattr(namespace, 'foo', False))
+ ]
+
+ - add_arguments(parser):
+
+ adds `--steps` and `--skip-steps` arguments to the parser. They can
+ be used to include or exclude steps during subcommand's execution.
+
+ - get_description(namespace):
+
+ collects steps' descriptions so that the subcommand description
+ is based on the descriptions of its steps.
+
+ - run(namespace):
+
+ implemented to resolve the steps to be run and to invoke them
+ in the order they are provided.
+
+ - has_dry_run and has_interactive_run are also set to True because usually
+ a steps based subcommand supports interactive and dry execution.
+"""
__metaclass__ = type
__all__ = [
- 'StepsBasedSubCommand',
+ 'add_common_arguments',
'ArgumentParser',
'BaseSubCommand',
+ 'get_args_from_namespace',
+ 'get_step_description',
+ 'get_step_name',
+ 'init_namespace',
+ 'resolve_steps',
+ 'restart_as_root',
+ 'StepsBasedSubCommand',
]
import argparse
import os
+import string
import subprocess
import sys
+import textwrap
-from lpsetup.exceptions import ValidationError
-from lpsetup.utils import (
- confirm,
- get_step_description,
- )
+from lpsetup.utils import get_terminal_width
class ArgumentParser(argparse.ArgumentParser):
@@ -28,7 +161,6 @@
def __init__(self, *args, **kwargs):
self.actions = []
- self.subparsers = None
super(ArgumentParser, self).__init__(*args, **kwargs)
def add_argument(self, *args, **kwargs):
@@ -43,213 +175,40 @@
action = super(ArgumentParser, self).add_argument(*args, **kwargs)
self.actions.append(action)
- def register_subcommand(self, name, subcommand_class, callback=None):
- """Add a subcommand to this parser.
-
- A sub command is registered giving the sub command `name` and class::
-
- >>> parser = ArgumentParser()
-
- >>> class SubCommand(BaseSubCommand):
- ... def run(self, namespace):
- ... return self.name
-
- >>> parser = ArgumentParser()
- >>> _ = parser.register_subcommand('foo', SubCommand)
-
- The `main` method of the subcommand class is added to namespace, and
- can be used to actually handle the sub command execution.
-
- >>> namespace = parser.parse_args(['foo'])
- >>> namespace.main(namespace)
- 'foo'
-
- A `callback` callable can also be provided to handle the subcommand
- execution::
-
- >>> callback = lambda namespace: 'custom callback'
-
- >>> parser = ArgumentParser()
- >>> _ = parser.register_subcommand(
- ... 'bar', SubCommand, callback=callback)
-
- >>> namespace = parser.parse_args(['bar'])
- >>> namespace.main(namespace)
- 'custom callback'
- """
- if self.subparsers is None:
- self.subparsers = self.add_subparsers(
- title='subcommands',
- help='Each subcommand accepts --h or --help to describe it.')
- subcommand = subcommand_class(name, callback=callback)
- parser = self.subparsers.add_parser(
- subcommand.name, help=subcommand.help)
- subcommand.add_arguments(parser)
- parser.set_defaults(main=subcommand.main, get_parser=lambda: parser)
- return subcommand
-
- def get_args_from_namespace(self, namespace):
- """Return a list of arguments taking values from `namespace`.
-
- Having a parser defined as usual::
-
- >>> parser = ArgumentParser()
- >>> _ = parser.add_argument('--foo')
- >>> _ = parser.add_argument('bar')
- >>> namespace = parser.parse_args('--foo eggs spam'.split())
-
- It is possible to recreate the argument list taking values from
- a different namespace::
-
- >>> namespace.foo = 'changed'
- >>> parser.get_args_from_namespace(namespace)
- ['--foo', 'changed', 'spam']
- """
- args = []
- for action in self.actions:
- dest = action.dest
- option_strings = action.option_strings
- value = getattr(namespace, dest, None)
- isbool = isinstance(value, bool)
- # If the value is boolean and the action is 'store_false', we
- # invert the value. This way the following `if value:` block
- # is executed if the original value is False, and the argument
- # is correctly added.
- if isbool and isinstance(action, argparse._StoreFalseAction):
- value = not value
- if value:
- if option_strings:
- args.append(option_strings[0])
- if isinstance(value, list):
- args.extend(value)
- elif not isbool:
- args.append(value)
- return args
-
- def _run_help(self, namespace):
- """Help sub command callback."""
- command = namespace.command
- help = self.prefix_chars + 'h'
- args = [help] if command is None else [command, help]
- self.parse_args(args)
-
- def _add_help_subcommand(self):
- """Add an help sub command to this parser."""
- name = 'help'
- choices = self.subparsers.choices.keys()
- if name not in choices:
- choices.append(name)
- parser = self.subparsers.add_parser(
- name, help='More help on a command.')
- parser.add_argument('command', nargs='?', choices=choices)
- parser.set_defaults(main=self._run_help)
-
- def parse_args(self, *args, **kwargs):
- """Override to add an help sub command.
-
- The help sub command is added only if other sub commands exist::
-
- >>> stderr, sys.stderr = sys.stderr, sys.stdout
- >>> parser = ArgumentParser()
- >>> parser.parse_args(['help'])
- Traceback (most recent call last):
- SystemExit: 2
- >>> sys.stderr = stderr
-
- >>> class SubCommand(BaseSubCommand): pass
- >>> _ = parser.register_subcommand('command', SubCommand)
- >>> namespace = parser.parse_args(['help'])
- >>> namespace.main(namespace)
- Traceback (most recent call last):
- SystemExit: 0
- """
- if self.subparsers is not None:
- self._add_help_subcommand()
- return super(ArgumentParser, self).parse_args(*args, **kwargs)
-
class BaseSubCommand(object):
- """Base class defining a sub command.
+ """Base class defining a subcommand.
Objects of this class can be used to easily add sub commands to this
- script as plugins, providing arguments, validating them, restarting
- as root if needed.
-
- Override `add_arguments()` to add arguments, `handlers` to add
- namespace handlers, and `run()` to manage sub command execution::
-
- >>> def handler(namespace):
- ... namespace.bar = True
-
- >>> class SubCommand(BaseSubCommand):
- ... help = 'Sub command example.'
- ... handlers = (handler,)
- ...
- ... def add_arguments(self, parser):
- ... super(SubCommand, self).add_arguments(parser)
- ... parser.add_argument('--foo')
- ...
- ... def run(self, namespace):
- ... return namespace
-
- Namespace handlers can be used to initialize and/or validate the
- arguments namespace.
-
- Register the sub command using `ArgumentParser.register_subcommand`::
-
- >>> parser = ArgumentParser()
- >>> sub_command = parser.register_subcommand('spam', SubCommand)
-
- Now the subcommand has a name::
-
- >>> sub_command.name
- 'spam'
-
- The sub command can be executed using `namespace.main()`::
-
- >>> namespace = parser.parse_args('spam --foo eggs'.split())
- >>> namespace = namespace.main(namespace)
- >>> namespace.foo
- 'eggs'
- >>> namespace.bar
- True
-
- The help attribute of sub command instances is used to generate
- the command usage message::
-
- >>> help = parser.format_help()
- >>> 'spam' in help
- True
- >>> 'Sub command example.' in help
- True
+ script as plug-ins, providing arguments and validating them.
+
+ See the docstring of this module for further details.
"""
help = ''
- needs_root = False
+ root_required = False
handlers = ()
- def __init__(self, name, callback=None):
- self.name = name
- self.callback = callback or self.run
-
def __repr__(self):
return '<{klass}: {name}>'.format(
- klass=self.__class__.__name__, name=self.name)
-
- def init_namespace(self, namespace):
- """Add `run_as_root` and `euid` names to the given `namespace`."""
- euid = os.geteuid()
- namespace.euid, namespace.run_as_root = euid, not euid
-
- def get_handlers(self, namespace):
- """Return an iterable of namespace handlers for this sub command.
-
- Subclasses can override this to dynamically change handlers
- based on the given `namespace`.
+ klass=self.__class__.__name__, name=self.help)
+
+ def add_arguments(self, parser):
+ """Here subclasses can add subcommand specific args to the parser."""
+ pass
+
+ def get_description(self, namespace):
+ return self.help
+
+ def needs_root(self, namespace):
+ """Return True if root is needed to run this subcommand.
+
+ Subclasses can override this to dynamically change this value
+ based on the given *namespace*.
"""
- return self.handlers
+ return self.root_required
- def prepare_namespace(self, parser, namespace):
+ def prepare_namespace(self, namespace):
"""Prepare the current namespace running namespace handlers.
The method `self.get_handlers` can return an iterable of objects
@@ -258,73 +217,25 @@
each other, or on the current environment.
Each handler is a callable object, takes the current namespace
- and can raise ValidationError if the arguments are not valid::
-
- >>> import sys
- >>> stderr, sys.stderr = sys.stderr, sys.stdout
- >>> def handler(namespace):
- ... raise ValidationError('nothing is going on')
- >>> sub_command = BaseSubCommand('foo')
- >>> sub_command.handlers = [handler]
- >>> sub_command.prepare_namespace(
- ... ArgumentParser(), argparse.Namespace())
- Traceback (most recent call last):
- SystemExit: 2
- >>> sys.stderr = stderr
+ and can raise ValidationError if the arguments are not valid.
"""
for handler in self.get_handlers(namespace):
- try:
- handler(namespace)
- except ValidationError as err:
- parser.error(err)
-
- def get_needs_root(self, namespace):
- """Return True if root is needed to run this subcommand.
-
- Subclasses can override this to dynamically change this value
- based on the given `namespace`.
- """
- return self.needs_root
-
- def restart_as_root(self, parser, namespace):
- """Restart this script using *sudo*.
-
- The arguments are recreated using the given `namespace`.
- """
- args = parser.get_args_from_namespace(namespace)
- return subprocess.call(['sudo', sys.argv[0], self.name] + args)
-
- def main(self, namespace):
- """Entry point for subcommand execution.
-
- This method takes care of:
-
- - current argparse subparser retrieval
- - namespace initialization
- - namespace handling/validation
- - script restart as root (if this sub command needs to be run as root)
-
- If everything is ok the sub command callback is called passing
- the initialized/validated namespace.
- """
- parser = namespace.get_parser()
- self.init_namespace(namespace)
- self.prepare_namespace(parser, namespace)
- if self.get_needs_root(namespace) and not namespace.run_as_root:
- return self.restart_as_root(parser, namespace)
- return self.callback(namespace)
+ handler(namespace)
def run(self, namespace):
- """Default sub command callback.
+ """Default subcommand callback.
- Subclasses must either implement this method or provide another
- callback during sub command registration.
+ This method must be implemented by subclasses.
"""
raise NotImplementedError
- def add_arguments(self, parser):
- """Here subclasses can add arguments to the subparser."""
- pass
+ def get_handlers(self, namespace):
+ """Return an iterable of namespace handlers for this sub command.
+
+ Subclasses can override this to dynamically change handlers
+ based on the given *namespace*.
+ """
+ return self.handlers
class StepsBasedSubCommand(BaseSubCommand):
@@ -345,8 +256,8 @@
>>> class SubCommand(StepsBasedSubCommand):
... has_interactive_run = False
... steps = (
- ... (step1, 'foo'),
- ... (step2, 'foo', 'bar'),
+ ... (step1, ['foo']),
+ ... (step2, ['foo', 'bar']),
... )
...
... def add_arguments(self, parser):
@@ -355,12 +266,13 @@
... parser.add_argument('--bar')
This class implements a `run` method that executes steps in the
- order they are provided::
+ order they are provided.
+ >>> subcommand = SubCommand()
>>> parser = ArgumentParser()
- >>> _ = parser.register_subcommand('sub', SubCommand)
- >>> namespace = parser.parse_args('sub --foo eggs --bar spam'.split())
- >>> namespace.main(namespace)
+ >>> subcommand.add_arguments(parser)
+ >>> namespace = parser.parse_args('--foo eggs --bar spam'.split())
+ >>> subcommand.run(namespace)
>>> trace
['step1 received eggs', 'step2 received eggs and spam']
@@ -369,8 +281,8 @@
>>> trace = []
- >>> namespace = parser.parse_args('sub --foo eggs -s step1'.split())
- >>> namespace.main(namespace)
+ >>> namespace = parser.parse_args('--foo eggs -s step1'.split())
+ >>> subcommand.run(namespace)
>>> trace
['step1 received eggs']
@@ -380,173 +292,175 @@
>>> trace = []
>>> namespace = parser.parse_args(
- ... 'sub --foo eggs --skip-steps step1'.split())
- >>> namespace.main(namespace)
+ ... '--foo eggs --skip-steps step1'.split())
+ >>> subcommand.run(namespace)
>>> trace
['step2 received eggs and None']
-
- The steps' execution is stopped if a step raises an exception.
-
- >>> trace = []
-
- >>> def erroneous_step(foo):
- ... raise subprocess.CalledProcessError(1, 'command')
-
- >>> class SubCommandWithErrors(SubCommand):
- ... steps = (
- ... (step1, 'foo'),
- ... (erroneous_step, 'foo'),
- ... (step2, 'foo', 'bar'),
- ... )
-
- >>> parser = ArgumentParser()
- >>> _ = parser.register_subcommand('sub', SubCommandWithErrors)
- >>> namespace = parser.parse_args('sub --foo eggs'.split())
- >>> namespace.main(namespace)
- Traceback (most recent call last):
- CalledProcessError: Command 'command' returned non-zero exit status 1
-
- The step `step2` is not executed::
-
- >>> trace
- ['step1 received eggs']
-
- A dynamic dispatcher takes care of running the steps. This way, a
- sub command can easily wrap or override the provided steps, defining a
- method named 'call_[step name]'::
-
- >>> trace = []
-
- >>> class DynamicSubCommand(SubCommand):
- ... def call_step1(self, namespace, step, args):
- ... trace.append('running step')
- ... step(*args)
-
- >>> parser = ArgumentParser()
- >>> _ = parser.register_subcommand('sub', DynamicSubCommand)
- >>> namespace = parser.parse_args('sub --foo eggs --bar spam'.split())
- >>> namespace.main(namespace)
- >>> trace
- ['running step', 'step1 received eggs', 'step2 received eggs and spam']
"""
steps = ()
+ has_dry_run = True
has_interactive_run = True
def add_arguments(self, parser):
- super(StepsBasedSubCommand, self).add_arguments(parser)
- # Add steps related arguments.
- step_names = [self._get_step_name(i[0]) for i in self.steps]
+ """Add steps related arguments."""
+ step_names = [get_step_name(i[0]) for i in self.steps]
parser.add_argument(
'-s', '--steps', nargs='+', choices=step_names,
help='Call one or more internal functions.')
parser.add_argument(
'--skip-steps', nargs='+', choices=step_names,
help='Skip one or more internal functions.')
- # Add developer interaction related arguments.
- if self.has_interactive_run:
- parser.add_argument(
- '-y', '--yes', action='store_false', dest='interactive',
- help='Assume yes to all queries.')
- parser.add_argument(
- '--dry-run', action='store_true', help='Dry run.')
-
- def init_namespace(self, namespace):
- """Override to add *namespace.interactive*.
-
- This is done to ensure *interactive* is always present as an
- attribute of *namespace*, even if self.has_interactive_run is False.
- """
- super(StepsBasedSubCommand, self).init_namespace(namespace)
- if not self.has_interactive_run:
- namespace.interactive = False
-
- def _get_step_name(self, step):
- """Return the string representation of a step callable.
-
- The name is retrieved using attributes lookup for `step_name`
- and then `__name__`::
-
- >>> def step1():
- ... pass
- >>> step1.step_name = 'mystep'
-
- >>> def step2():
- ... pass
-
- >>> sub_command = StepsBasedSubCommand('foo')
- >>> sub_command._get_step_name(step1)
- 'mystep'
- >>> sub_command._get_step_name(step2)
- 'step2'
- """
- try:
- return step.step_name
- except AttributeError:
- return step.__name__
-
- def _include_step(self, step_name, namespace):
- """Return True if the given *step_name* must be run, False otherwise.
-
- A step is included in the command execution if the step included in
- `--steps` (or steps is None) and is not included in `--skip-steps`.
- """
- steps_to_skip = namespace.skip_steps or []
- steps_to_run = namespace.steps
- # If explicitly told to skip a step, then skip it.
- if step_name in steps_to_skip:
- return False
- # If no list of steps was provided then any non-skipped are to be run.
- if steps_to_run is None:
- return True
- # A list of steps to run was provided, so the step has to be included
- # in order to be run.
- return step_name in steps_to_run
-
- def get_steps(self, namespace):
- """Return a list of *(step_name, step_callable, step_args)* tuples."""
- steps = []
- for step_arg_names in self.steps:
- step, arg_names = step_arg_names[0], step_arg_names[1:]
- step_name = self._get_step_name(step)
- if self._include_step(step_name, namespace):
- args = [getattr(namespace, i) for i in arg_names]
- steps.append((step_name, step, args))
- return steps
-
- def _call_step(self, namespace, step, args):
- """Default callable used to run a `step`, using given `args`."""
- return step(*args)
-
- def get_steps_description(self, namespace, steps):
- """Retrieve steps' descriptions from the given *steps*.
-
- Return a string containing all the descriptions.
- """
+
+ def get_description(self, namespace):
+ """Collect steps' descriptions and return them as a single string."""
context = namespace.__dict__
- descriptions = [get_step_description(i[1], **context) for i in steps]
+ steps = resolve_steps(self.steps, namespace)
+ descriptions = [get_step_description(i[0], **context) for i in steps]
return '\n'.join(filter(None, descriptions))
def run(self, namespace):
- steps = self.get_steps(namespace)
- if namespace.dry_run or namespace.interactive:
- # Collect and display the description of each step.
- description = self.get_steps_description(namespace, steps)
- if description:
- print 'This command will perform the following actions:\n'
- print description + '\n'
- # Quit without errors if this is a dry run.
- if namespace.dry_run:
- return
- # If this is not a dry run, then it is an interactive one.
- # Prompt the user for confirmation to proceed and quit if
- # requested (in this case with exit code 1).
- if not confirm():
- return 1
- # Execute all the steps.
- default_step_runner = self._call_step
- for step_name, step, args in steps:
- # Run the step using a dynamic dispatcher.
- step_runner = getattr(
- self, 'call_' + step_name, default_step_runner)
- step_runner(namespace, step, args)
+ """Execute the steps in the order they are provided."""
+ for step, args in resolve_steps(self.steps, namespace):
+ step(*args)
+
+
+def add_common_arguments(
+ parser, has_dry_run=False, has_interactive_run=False):
+ """Add to given *parser* the arguments that all subcommands have in common.
+
+ The common arguments are:
+
+ - `-y` or `--yes`: non-interactive execution.
+ - `--dry-run`: dry execution.
+
+ Ensure the resulting namespace always contains *interactive* and *dry_run*
+ names, even if *has_interactive_run* or *has_dry_run* are False.
+ """
+ if has_dry_run:
+ parser.add_argument('--dry-run', action='store_true', help='Dry run.')
+ else:
+ parser.set_defaults(dry_run=False)
+ if has_interactive_run:
+ parser.add_argument(
+ '-y', '--yes', action='store_false', dest='interactive',
+ help='Assume yes to all queries.')
+ else:
+ parser.set_defaults(interactive=False)
+
+
+def get_args_from_namespace(parser, namespace):
+ """Return a list of arguments taking values from `namespace`."""
+ args = []
+ for action in parser.actions:
+ dest = action.dest
+ option_strings = action.option_strings
+ value = getattr(namespace, dest, None)
+ isbool = isinstance(value, bool)
+ # If the value is boolean and the action is 'store_false', we
+ # invert the value. This way the following `if value:` block
+ # is executed if the original value is False, and the argument
+ # is correctly added.
+ if isbool and isinstance(action, argparse._StoreFalseAction):
+ value = not value
+ if value:
+ if option_strings:
+ args.append(option_strings[0])
+ if isinstance(value, list):
+ args.extend(value)
+ elif not isbool:
+ args.append(value)
+ return args
+
+
+def init_namespace(namespace):
+ """Add `run_as_root` and `euid` names to the given *namespace*."""
+ euid = os.geteuid()
+ namespace.euid, namespace.run_as_root = euid, not euid
+
+
+def restart_as_root(parser, namespace):
+ """Restart this script using *sudo*.
+
+ The arguments are recreated using the given `namespace`.
+ """
+ cmd = ['sudo', sys.argv[0]] + parser.prog.split()[1:]
+ namespace.interactive = False
+ args = get_args_from_namespace(parser, namespace)
+ return subprocess.call(cmd + args)
+
+
+def get_step_name(step):
+ """Return a string representing the name of the given *step*."""
+ try:
+ return step.step_name
+ except AttributeError:
+ return step.__name__
+
+
+def resolve_steps(steps, namespace):
+ """Return a list of *(step_callable, step_args)* tuples.
+
+ The argument *steps* is a sequence of
+ *(step_callable, arg_names, filter_callable)*.
+
+ The *arg_names* are strings representing arguments in the namespace.
+ The *filter_callable* is optional. It must take the namespace and return
+ True if the corresponding step must be included, False otherwise.
+
+ If the namespace contains the `steps` or `skip_steps` names, they are
+ taken into consideration to resolve the steps that must be run.
+
+ In essence, a step is returned as part of the sequence if:
+
+ - is not included in *namespace.skip_steps*
+ - is not excluded from *namespace.step* (if *namespace.step* is not
+ defined or is None, all the steps are considered as included)
+ - the provided filter callable returns True (if not present, the
+ step is included by default)
+ """
+ steps_to_skip = getattr(namespace, 'skip_steps', None) or []
+ steps_to_run = getattr(namespace, 'steps', None)
+ step_args = []
+ for step in steps:
+ step_callable, arg_names = step[:2]
+ step_name = get_step_name(step_callable)
+ # Include step if is not explicitly told to skip it.
+ if step_name not in steps_to_skip:
+ # If no list of steps was provided then any non-skipped are to be
+ # run, otherwise the step has to be included in order to be run.
+ if (steps_to_run is None) or (step_name in steps_to_run):
+ # The step filter function, if present, can deny the execution.
+ granted = step[2](namespace) if len(step) == 3 else True
+ if granted:
+ args = [getattr(namespace, i) for i in arg_names]
+ step_args.append((step_callable, args))
+ return step_args
+
+
+def get_step_description(step, **kwargs):
+ """Retrieve and format step description from the given *step* callable.
+
+ *kwargs*, if provided, will be used as context to format the description.
+ Formatting is done using the Python's built-in templating system
+ *string.Template* supporting $-based substitutions.
+
+ If placeholders are missing from *kwargs* an error will be raised.
+ """
+ description = step.description
+ if not description:
+ # This step has no description, nothing else to do.
+ return ''
+ if kwargs:
+ s = string.Template(description)
+ description = s.substitute(**kwargs)
+ # Remove multiple spaces from lines.
+ lines = [' '.join(line.split()) for line in description.splitlines()]
+ # Retrieve all the non empty lines.
+ lines = filter(None, lines)
+ # For each line, wrap the contents. Note that we can't wrap the text of
+ # the entire paragraph because we want to preserve existing new lines.
+ width = get_terminal_width()
+ return '\n'.join(textwrap.fill(
+ line, width=width, initial_indent='* ', subsequent_indent=' ')
+ for line in lines)
=== modified file 'lpsetup/cli.py'
--- lpsetup/cli.py 2012-07-09 15:19:13 +0000
+++ lpsetup/cli.py 2012-08-09 15:36:20 +0000
@@ -6,7 +6,10 @@
__metaclass__ = type
__all__ = [
+ 'get_parser',
'main',
+ 'prepare_parser',
+ 'run',
]
import sys
@@ -24,30 +27,144 @@
update,
version,
)
-
-
-subcommands = [
- ('finish-init-host', finish_inithost.SubCommand),
- ('init-host', inithost.SubCommand),
- ('init-lxc', initlxc.SubCommand),
- ('init-repo', initrepo.SubCommand),
- ('install-lxc', install_lxc.SubCommand),
- ('update', update.SubCommand),
- ('version', version.SubCommand),
+from lpsetup.utils import confirm
+
+
+SUBCOMMANDS = [
+ ('finish-init-host', finish_inithost.SubCommand()),
+ ('init-host', inithost.SubCommand()),
+ ('init-lxc', initlxc.SubCommand()),
+ ('init-repo', initrepo.SubCommand()),
+ ('install-lxc', install_lxc.SubCommand()),
+ ('update', update.SubCommand()),
+ ('version', version.SubCommand()),
]
-def main(args=None):
+def run(parser, subcommand, namespace):
+ """Run the *subcommand* using the given *namespace*.
+
+ This function takes care of::
+
+ - namespace initialization and validation
+ - interactive arguments handling
+ - executing a dry run, if requested
+ - restarting the same sub command as root, if required
+ - actual sub command execution
+ """
+ # Initialize the namespace.
+ argparser.init_namespace(namespace)
+
+ # Prepare the namespace. Each sub command can define a *prepare_namespace*
+ # method that can raise a ValidationError if the namespace is not valid.
+ try:
+ subcommand.prepare_namespace(namespace)
+ except exceptions.ValidationError as err:
+ parser.error(err)
+ except AttributeError:
+ pass
+
+ # Handle user interaction and dry run.
+ if namespace.dry_run or namespace.interactive:
+ # Display the sub command description.
+ command_description = subcommand.get_description(namespace)
+ if command_description:
+ print 'This command will perform the following actions:\n'
+ print command_description + '\n'
+ # Quit without errors if this is a dry run.
+ if namespace.dry_run:
+ return
+ # If this is not a dry run, then it is an interactive one.
+ # Prompt the user for confirmation to proceed and quit if
+ # requested (in this case with exit code 1).
+ try:
+ proceed = confirm()
+ except KeyboardInterrupt:
+ print '\nQuitting.'
+ return 1
+ if not proceed:
+ return 1
+
+ # Restart as root if needed.
+ if not namespace.run_as_root and subcommand.needs_root(namespace):
+ return argparser.restart_as_root(parser, namespace)
+
+ # Execute the command.
+ try:
+ return subcommand.run(namespace)
+ except exceptions.ExecutionError as err:
+ return err
+
+
+def prepare_parser(parser, subcommand):
+ """Add common and sub command specific arguments to the given *parser*.
+
+ Also add to the parser defaults a *main* function that, taking a parsed
+ namespace, executes the current sub command with the current parser.
+ """
+ def main(namespace):
+ return run(parser, subcommand, namespace)
+
+ # Store a reference to the current sub command entry point.
+ parser.set_defaults(main=main)
+ # Add user interaction related arguments.
+ argparser.add_common_arguments(
+ parser,
+ has_dry_run=getattr(subcommand, 'has_dry_run', False),
+ has_interactive_run=getattr(subcommand, 'has_interactive_run', False),
+ )
+ # Add sub command specific arguments, if requested.
+ try:
+ subcommand.add_arguments(parser)
+ except AttributeError:
+ pass
+
+
+def get_parser(subcommands):
+ """Return the argument parser with all subcommands registered.
+
+ Also add the help subcommand.
+ """
+ parser = argparser.ArgumentParser(description=description)
+ subparsers = parser.add_subparsers(
+ title='subcommands',
+ help='Each subcommand accepts --h or --help to describe it.')
+ for name, subcommand in subcommands:
+ # Register the sub command.
+ help = subcommand.help
+ subparser = subparsers.add_parser(name, description=help, help=help)
+ # Prepare the sub command arguments.
+ prepare_parser(subparser, subcommand)
+
+ # Add the *help* sub command.
+ def main(namespace):
+ command = namespace.command
+ help = parser.prefix_chars + 'h'
+ args = [help] if command is None else [command, help]
+ parser.parse_args(args)
+
+ choices = dict(subcommands).keys()
+ help = 'More help on a command.'
+ subparser = subparsers.add_parser('help', description=help, help=help)
+ subparser.add_argument('command', nargs='?', choices=choices)
+ subparser.set_defaults(main=main)
+
+ return parser
+
+
+def main(args=None, parser=None):
+ # Retrieve the command line arguments.
if args is None:
args = sys.argv[1:]
- parser = argparser.ArgumentParser(description=description)
- for name, klass in subcommands:
- parser.register_subcommand(name, klass)
+ # Get the parser.
+ if parser is None:
+ parser = get_parser(SUBCOMMANDS)
+ # Parse the command line arguments.
try:
- args = parser.parse_args(args)
+ namespace = parser.parse_args(args)
+ # Run the sub command. Each subparser adds to namespace a main
+ # function that basically execute run() passing the corresponding
+ # sub command, sub parser and the namespace itself.
+ return namespace.main(namespace)
except SystemExit as err:
return err.code
- try:
- return args.main(args)
- except (exceptions.ExecutionError, KeyboardInterrupt) as err:
- return err
=== modified file 'lpsetup/subcommands/finish_inithost.py'
--- lpsetup/subcommands/finish_inithost.py 2012-08-01 14:42:01 +0000
+++ lpsetup/subcommands/finish_inithost.py 2012-08-09 15:36:20 +0000
@@ -69,13 +69,10 @@
"""
help = __doc__
-
- needs_root = True
-
+ root_required = True
steps = (
- (setup_launchpad, 'user', 'target_dir'),
+ (setup_launchpad, ['user', 'target_dir']),
)
-
handlers = (handle_user, handle_target_dir)
def add_arguments(self, parser):
=== modified file 'lpsetup/subcommands/inithost.py'
--- lpsetup/subcommands/inithost.py 2012-08-08 20:25:02 +0000
+++ lpsetup/subcommands/inithost.py 2012-08-09 15:36:20 +0000
@@ -8,7 +8,9 @@
__all__ = [
'initialize',
'initialize_base',
+ 'initialize_lxc',
'setup_apt',
+ 'setup_home',
'SubCommand',
]
@@ -214,7 +216,7 @@
def initialize_lxc():
"""Initialize LXC container.
- Note that this step is guarded by the call_initialize_lxc method so that
+ Note that this step is guarded by a filter callable method so that
it is only called when lpsetup is running in a container.
"""
lxc_os = get_distro()
@@ -263,15 +265,16 @@
class SubCommand(argparser.StepsBasedSubCommand):
"""Prepare a machine to run Launchpad. May be an LXC container or not."""
- initialize_step = (initialize, 'user')
-
- setup_home_step = (setup_home,
- 'user', 'full_name', 'email', 'lpuser',
- 'valid_ssh_keys', 'ssh_key_path')
-
- initialize_lxc_step = (initialize_lxc, )
-
- setup_apt_step = (setup_apt, )
+ initialize_step = (initialize, ['user'])
+
+ setup_home_step = (
+ setup_home, ['user', 'full_name', 'email', 'lpuser',
+ 'valid_ssh_keys', 'ssh_key_path'])
+
+ initialize_lxc_step = (
+ initialize_lxc, [], lambda namespace: running_in_container())
+
+ setup_apt_step = (setup_apt, [])
steps = (
initialize_step,
@@ -281,7 +284,7 @@
)
help = __doc__
- needs_root = True
+ root_required = True
handlers = (
handle_user,
handle_lpuser_as_username,
@@ -289,11 +292,6 @@
handle_ssh_keys,
)
- def call_initialize_lxc(self, namespace, step, args):
- """Caller that only initializes LXC if we are in an LXC."""
- if running_in_container():
- return step(*args)
-
def add_arguments(self, parser):
super(SubCommand, self).add_arguments(parser)
parser.add_argument(
=== modified file 'lpsetup/subcommands/initlxc.py'
--- lpsetup/subcommands/initlxc.py 2012-08-08 20:25:02 +0000
+++ lpsetup/subcommands/initlxc.py 2012-08-09 15:36:20 +0000
@@ -12,6 +12,7 @@
__all__ = [
'create_lxc',
'inithost_in_lxc',
+ 'initialize',
'install_lpsetup_in_lxc',
'start_lxc',
'stop_lxc',
@@ -224,7 +225,7 @@
if not lxc_stopped(lxc_name):
subprocess.call(['lxc-stop', '-n', lxc_name])
-stop_lxc.description = None
+stop_lxc.description = 'Stop the LXC instance $lxc_name.\n'
class SubCommand(inithost.SubCommand):
@@ -232,23 +233,25 @@
development environment.
"""
- create_lxc_step = (create_lxc,
- 'lxc_name', 'lxc_arch', 'lxc_os', 'user', 'install_subunit')
- start_lxc_step = (start_lxc,
- 'lxc_name')
- wait_for_lxc_step = (wait_for_lxc,
- 'lxc_name', 'ssh_key_path')
- install_lpsetup_in_lxc_step = (install_lpsetup_in_lxc,
- 'lxc_name', 'ssh_key_path', 'lxc_os', 'user', 'home_dir',
- 'lpsetup_branch')
- inithost_in_lxc_step = (inithost_in_lxc,
- 'lxc_name', 'ssh_key_path', 'user', 'email', 'full_name', 'lpuser',
- 'ssh_key_name', 'home_dir')
- stop_lxc_step = (stop_lxc,
- 'lxc_name', 'ssh_key_path')
+ create_lxc_step = (
+ create_lxc,
+ ['lxc_name', 'lxc_arch', 'lxc_os', 'user', 'install_subunit'])
+ start_lxc_step = (start_lxc, ['lxc_name'])
+ wait_for_lxc_step = (wait_for_lxc, ['lxc_name', 'ssh_key_path'])
+ install_lpsetup_in_lxc_step = (
+ install_lpsetup_in_lxc,
+ ['lxc_name', 'ssh_key_path', 'lxc_os', 'user', 'home_dir',
+ 'lpsetup_branch'])
+ inithost_in_lxc_step = (
+ inithost_in_lxc,
+ ['lxc_name', 'ssh_key_path', 'user', 'email', 'full_name', 'lpuser',
+ 'ssh_key_name', 'home_dir'])
+ stop_lxc_step = (
+ stop_lxc, ['lxc_name', 'ssh_key_path'],
+ lambda namespace: namespace.stop_lxc)
base_steps = (
- (initialize, 'user', 'install_haveged'),
+ (initialize, ['user', 'install_haveged']),
inithost.SubCommand.setup_home_step,
create_lxc_step,
start_lxc_step,
@@ -259,12 +262,7 @@
steps = base_steps + (stop_lxc_step,)
help = __doc__
- needs_root = True
-
- def call_stop_lxc(self, namespace, step, args):
- """Run the `stop_lxc` step only if the related flag is set."""
- if namespace.stop_lxc:
- return step(*args)
+ root_required = True
def add_arguments(self, parser):
super(SubCommand, self).add_arguments(parser)
=== modified file 'lpsetup/subcommands/initrepo.py'
--- lpsetup/subcommands/initrepo.py 2012-08-08 20:25:02 +0000
+++ lpsetup/subcommands/initrepo.py 2012-08-09 15:36:20 +0000
@@ -15,6 +15,8 @@
__metaclass__ = type
__all__ = [
+ 'fetch',
+ 'setup_bzr_locations',
'SubCommand',
]
@@ -93,8 +95,8 @@
lpuser, repository, branch_name, template=LP_BZR_LOCATIONS):
"""Set up bazaar locations.
- Note that this step is guarded by the *call_setup_bzr_locations* method
- so that it is only called when the user Launchpad login can be retrieved.
+ Note that this step is guarded by a filter callable so that it is only
+ called when the user Launchpad login can be retrieved.
"""
context = {
'branch_dir': os.path.join(repository, branch_name),
@@ -119,8 +121,8 @@
f.write(wrap_contents(contents.getvalue()))
contents.close()
-setup_bzr_locations.description = """If bzr+ssh is used, update bazaar \
- locations ($home_dir/.bazaar/locations.conf) to include repository \
+setup_bzr_locations.description = """Update bazaar locations \
+ ($home_dir/.bazaar/locations.conf) to include repository \
$repository and branch $branch_name.
"""
@@ -130,9 +132,10 @@
steps = (
(fetch,
- 'source', 'repository', 'branch_name', 'checkout_name',
- 'no_checkout'),
- (setup_bzr_locations, 'lpuser', 'repository', 'branch_name'),
+ ['source', 'repository', 'branch_name', 'checkout_name',
+ 'no_checkout']),
+ (setup_bzr_locations, ['lpuser', 'repository', 'branch_name'],
+ lambda namespace: namespace.lpuser is not None),
)
help = __doc__
@@ -143,11 +146,7 @@
handlers.handle_branch_and_checkout,
handlers.handle_source,
)
-
- def call_setup_bzr_locations(self, namespace, step, args):
- """Caller that only sets up bzr locations if lpuser exists."""
- if namespace.lpuser is not None:
- return step(*args)
+ root_required = False
@staticmethod
def add_common_arguments(parser):
=== modified file 'lpsetup/subcommands/install_lxc.py'
--- lpsetup/subcommands/install_lxc.py 2012-08-09 12:15:53 +0000
+++ lpsetup/subcommands/install_lxc.py 2012-08-09 15:36:20 +0000
@@ -7,7 +7,10 @@
__metaclass__ = type
__all__ = [
'create_scripts',
+ 'finish_inithost_in_lxc',
+ 'init_repo_in_lxc',
'SubCommand',
+ 'update_in_lxc',
]
import os
@@ -69,7 +72,7 @@
with open(procfile, 'w') as f:
f.write('0\n')
-create_scripts.description = """If requested, create helper script \
+create_scripts.description = """Create helper scripts \
/usr/local/bin/lp-setup-*: they can be used to build Launchpad and \
start a parallel test run.
"""
@@ -125,25 +128,24 @@
class SubCommand(initlxc.SubCommand):
- """Completely sets up an LXC environment with Launchpad using the sandbox
- development model.
- """
+ """Completely sets up an LXC environment with Launchpad."""
steps = initlxc.SubCommand.base_steps + (
# Run on host:
- (create_scripts, 'lxc_name', 'ssh_key_path', 'user'),
+ (create_scripts, ['lxc_name', 'ssh_key_path', 'user'],
+ lambda namespace: namespace.create_scripts),
# Run inside the container:
(init_repo_in_lxc,
- 'lxc_name', 'ssh_key_path', 'home_dir', 'user', 'source', 'use_http',
- 'branch_name', 'checkout_name', 'repository', 'no_checkout'),
- (update_in_lxc, 'lxc_name', 'ssh_key_path', 'home_dir', 'user',
- 'external_path', 'target_dir', 'lp_source_deps', 'use_http'),
- (finish_inithost_in_lxc, 'lxc_name', 'ssh_key_path', 'home_dir',
- 'user', 'target_dir'),
+ ['lxc_name', 'ssh_key_path', 'home_dir', 'user', 'source', 'use_http',
+ 'branch_name', 'checkout_name', 'repository', 'no_checkout']),
+ (update_in_lxc,
+ ['lxc_name', 'ssh_key_path', 'home_dir', 'user', 'external_path',
+ 'target_dir', 'lp_source_deps', 'use_http']),
+ (finish_inithost_in_lxc,
+ ['lxc_name', 'ssh_key_path', 'home_dir', 'user', 'target_dir']),
# Run on host:
initlxc.SubCommand.stop_lxc_step,
)
-
help = __doc__
def get_handlers(self, namespace):
@@ -156,11 +158,6 @@
handle_target_from_repository,
)
- def call_create_scripts(self, namespace, step, args):
- """Run the `create_scripts` step only if the related flag is set."""
- if namespace.create_scripts:
- return step(*args)
-
def add_arguments(self, parser):
super(SubCommand, self).add_arguments(parser)
# Inherit arguments from subcommands we depend upon.
=== modified file 'lpsetup/subcommands/update.py'
--- lpsetup/subcommands/update.py 2012-08-02 14:44:32 +0000
+++ lpsetup/subcommands/update.py 2012-08-09 15:36:20 +0000
@@ -6,7 +6,10 @@
__metaclass__ = type
__all__ = [
+ 'initialize_directories',
'SubCommand',
+ 'update_dependencies',
+ 'update_tree',
]
import os.path
@@ -83,10 +86,10 @@
has_interactive_run = False
steps = (
- (initialize_directories, 'target_dir', 'external_path'),
+ (initialize_directories, ['target_dir', 'external_path']),
(update_dependencies,
- 'target_dir', 'external_path', 'use_http', 'lp_source_deps'),
- (update_tree, 'target_dir'),
+ ['target_dir', 'external_path', 'use_http', 'lp_source_deps']),
+ (update_tree, ['target_dir']),
)
help = __doc__
handlers = (
=== removed file 'lpsetup/tests/examples.py'
--- lpsetup/tests/examples.py 2012-07-30 10:05:19 +0000
+++ lpsetup/tests/examples.py 1970-01-01 00:00:00 +0000
@@ -1,93 +0,0 @@
-#!/usr/bin/env python
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Example objects used in tests."""
-
-__metaclass__ = type
-__all__ = [
- 'step1',
- 'step2',
- 'StepsBasedSubCommand',
- 'StepsBasedSubCommandWithErrors',
- 'bad_step',
- 'SubCommand',
- ]
-
-import subprocess
-
-from lpsetup import argparser
-
-
-def step1(foo):
- print 'step1 received ' + foo
-
-
-def step2(foo, bar):
- print 'step2 received {0} and {1}'.format(foo, bar)
-
-step2.step_name = 'mystep'
-
-
-def step_with_description(foo):
- pass
-
-step_with_description.description = 'step description'
-
-
-def bad_step(foo):
- raise subprocess.CalledProcessError(1, 'command')
-
-
-class SubCommand(argparser.BaseSubCommand):
- """An example sub command."""
-
- help = 'Sub command example.'
-
- def add_arguments(self, parser):
- super(SubCommand, self).add_arguments(parser)
- parser.add_argument('--foo')
-
- def run(self, namespace):
- return namespace
-
-
-class StepsBasedSubCommand(argparser.StepsBasedSubCommand):
- """An example steps based sub command."""
-
- has_interactive_run = False
- steps = (
- (step1, 'foo'),
- (step2, 'foo', 'bar'),
- )
-
- def add_arguments(self, parser):
- super(StepsBasedSubCommand, self).add_arguments(parser)
- parser.add_argument('--foo')
- parser.add_argument('--bar')
-
-
-class StepsBasedSubCommandWithErrors(StepsBasedSubCommand):
- """An example steps based sub command (containing a failing step)."""
-
- steps = (
- (step1, 'foo'),
- (bad_step, 'foo'),
- (step2, 'foo', 'bar'),
- )
-
-
-class DynamicStepsBasedSubCommand(StepsBasedSubCommand):
- """An example steps based sub command (using internal step dispatcher)."""
-
- def call_step1(self, namespace, step, args):
- print 'running step1 with {args} while bar is {bar}'.format(
- args=','.join(args), bar=namespace.bar)
- step(*args)
-
-
-class InteractiveStepsBasedSubCommand(StepsBasedSubCommand):
- """An example interactive steps based sub command."""
-
- has_interactive_run = True
- steps = [(step_with_description, 'foo')]
=== modified file 'lpsetup/tests/subcommands/test_inithost.py'
--- lpsetup/tests/subcommands/test_inithost.py 2012-08-06 10:04:17 +0000
+++ lpsetup/tests/subcommands/test_inithost.py 2012-08-09 15:36:20 +0000
@@ -47,11 +47,7 @@
handlers.handle_userdata,
handlers.handle_ssh_keys,
)
- expected_steps = (
- initialize_step,
- setup_home_step,
- initialize_lxc_step,
- setup_apt_step,)
+ expected_steps = (initialize_step, setup_home_step, setup_apt_step)
needs_root = True
=== modified file 'lpsetup/tests/subcommands/test_initrepo.py'
--- lpsetup/tests/subcommands/test_initrepo.py 2012-07-25 15:57:26 +0000
+++ lpsetup/tests/subcommands/test_initrepo.py 2012-08-09 15:36:20 +0000
@@ -48,7 +48,6 @@
@skip_if_no_lpuser
class InitrepoTest(StepsBasedSubCommandTestMixin, unittest.TestCase):
- sub_command_name = 'init-repo'
sub_command_class = initrepo.SubCommand
expected_arguments = get_arguments()
expected_handlers = (
=== modified file 'lpsetup/tests/subcommands/test_smoke.py'
--- lpsetup/tests/subcommands/test_smoke.py 2012-07-30 13:37:25 +0000
+++ lpsetup/tests/subcommands/test_smoke.py 2012-08-09 15:36:20 +0000
@@ -7,13 +7,10 @@
from lpsetup.cli import (
main,
- subcommands,
+ SUBCOMMANDS,
)
-from lpsetup.tests.utils import (
- capture_output,
- root_not_needed,
- )
+from lpsetup.tests.utils import capture_output
class SmokeTest(unittest.TestCase):
@@ -22,8 +19,8 @@
def test_subcommand_smoke_via_help(self):
# Perform a basic smoke test by running each subcommand's --help
# function and verify that a non-error exit code is returned.
- for subcommand, callable in subcommands:
- self.assertEqual(main([subcommand, '--help']), 0)
+ for name, _ in SUBCOMMANDS:
+ self.assertEqual(main([name, '--help']), 0)
def test_subcommand_smoke_via_dry_run(self):
# Perform a basic smoke test by running each subcommand's dry run
@@ -37,13 +34,10 @@
'install-lxc': required_args,
'update': [],
}
- name_class_map = dict(subcommands)
warning = 'This command will perform the following actions:'
for name, args in name_args_map.items():
- subcommand = name_class_map[name]
with capture_output() as output:
- with root_not_needed(subcommand):
- retcode = main([name, '--dry-run'] + args)
+ retcode = main([name, '--dry-run'] + args)
self.assertFalse(retcode)
# Ensure sub command description is printed to stdout.
self.assertIn(warning, output.getvalue())
=== modified file 'lpsetup/tests/subcommands/test_update.py'
--- lpsetup/tests/subcommands/test_update.py 2012-08-01 16:17:15 +0000
+++ lpsetup/tests/subcommands/test_update.py 2012-08-09 15:36:20 +0000
@@ -33,7 +33,6 @@
class UpdateTest(StepsBasedSubCommandTestMixin, unittest.TestCase):
- sub_command_name = 'update'
sub_command_class = update.SubCommand
expected_arguments = get_arguments()
expected_handlers = (handlers.handle_user, handlers.handle_target_dir)
=== modified file 'lpsetup/tests/subcommands/test_version.py'
--- lpsetup/tests/subcommands/test_version.py 2012-06-25 10:41:46 +0000
+++ lpsetup/tests/subcommands/test_version.py 2012-08-09 15:36:20 +0000
@@ -4,22 +4,24 @@
"""Tests for the version sub command."""
+import argparse
+import os
+import sys
import unittest
from lpsetup import get_version
from lpsetup.subcommands import version
-from lpsetup.tests.utils import (
- capture_output,
- SubCommandTestMixin,
- )
-
-
-class VersionTest(SubCommandTestMixin, unittest.TestCase):
-
- sub_command_class = version.SubCommand
+from lpsetup.tests.utils import capture_output
+
+
+class VersionTest(unittest.TestCase):
def test_sub_command(self):
+ # Ensure the version sub command prints to stdout
+ # the actual app name and version.
+ subcommand = version.SubCommand()
with capture_output() as output:
- self.parse_and_call_main()
- bits = output.getvalue().strip().split()
- self.assertEqual('v' + get_version(), bits[1])
+ subcommand.run(argparse.Namespace())
+ app_name, app_version = output.getvalue().strip().split()
+ self.assertEqual(os.path.basename(sys.argv[0]), app_name)
+ self.assertEqual('v' + get_version(), app_version)
=== modified file 'lpsetup/tests/test_argparser.py'
--- lpsetup/tests/test_argparser.py 2012-07-30 12:04:54 +0000
+++ lpsetup/tests/test_argparser.py 2012-08-09 15:36:20 +0000
@@ -4,30 +4,69 @@
"""Tests for the argparser module."""
-from contextlib import nested
-import subprocess
+import argparse
+from collections import OrderedDict
+import itertools
+import sys
import unittest
from lpsetup import argparser
+from lpsetup.utils import get_terminal_width
from lpsetup.exceptions import ValidationError
-from lpsetup.tests import examples
from lpsetup.tests.utils import (
- capture_error,
- capture_output,
- RawInputReturning,
- SubCommandTestMixin,
+ call_replaced_by,
+ ParserTestMixin,
+ StepsFactoryTestMixin,
+ SubCommandFactoryTestMixin,
)
-class ArgumentParserTest(unittest.TestCase):
-
- def setUp(self):
- self.parser = argparser.ArgumentParser()
-
- def get_sub_command(self):
- return type(
- 'SubCommand', (argparser.BaseSubCommand,),
- {'run': lambda self, namespace: self.name})
+class AddCommonArgumentsTest(ParserTestMixin, unittest.TestCase):
+
+ def test_interactive_supported(self):
+ # Ensure the resulting *namespace.interactive* is True if the sub
+ # command is interactive and `--yes` is not provided.
+ argparser.add_common_arguments(self.parser, has_interactive_run=True)
+ namespace = self.parser.parse_args([])
+ self.assertTrue(namespace.interactive)
+
+ def test_interactive_not_supported(self):
+ # Ensure the resulting *namespace.interactive* is False if the sub
+ # command does not support interactive execution.
+ argparser.add_common_arguments(self.parser, has_interactive_run=False)
+ namespace = self.parser.parse_args([])
+ self.assertFalse(namespace.interactive)
+
+ def test_assume_yes_provided(self):
+ # Ensure the resulting *namespace.interactive* is False if the sub
+ # command is interactive but `--yes` is provided.
+ argparser.add_common_arguments(self.parser, has_interactive_run=True)
+ namespace = self.parser.parse_args(['--yes'])
+ self.assertFalse(namespace.interactive)
+
+ def test_dry_run_supported(self):
+ # Ensure the resulting *namespace.dry_run* is False if the sub
+ # command supports dry execution but `--dry-run` is not provided.
+ argparser.add_common_arguments(self.parser, has_dry_run=True)
+ namespace = self.parser.parse_args([])
+ self.assertFalse(namespace.dry_run)
+
+ def test_dry_run_not_supported(self):
+ # Ensure the resulting *namespace.dry_run* is False if the sub
+ # command does not support dry execution.
+ argparser.add_common_arguments(self.parser, has_dry_run=False)
+ namespace = self.parser.parse_args([])
+ self.assertFalse(namespace.dry_run)
+
+ def test_dry_run_provided(self):
+ # Ensure the resulting *namespace.dry_run* is True if the sub
+ # command supports dry execution and `--dry-run` is provided.
+ argparser.add_common_arguments(self.parser, has_dry_run=True)
+ namespace = self.parser.parse_args(['--dry-run'])
+ self.assertTrue(namespace.dry_run)
+
+
+class ArgumentParserTest(ParserTestMixin, unittest.TestCase):
def test_add_argument(self):
# Ensure actions are stored in a "public" instance attribute.
@@ -36,228 +75,294 @@
dests = [action.dest for action in self.parser.actions]
self.assertListEqual(['help', 'arg1', 'arg2'], dests)
- def test_register_subcommand(self):
- # The `main` method of the subcommand class is added to namespace,
- # and can be used to actually handle the sub command execution.
- name = 'foo'
- self.parser.register_subcommand(name, self.get_sub_command())
- namespace = self.parser.parse_args([name])
- self.assertEqual(name, namespace.main(namespace))
-
- def test_register_subcommand_providing_handler(self):
- # Ensure a `handler` callable can also be provided to handle the
- # subcommand execution.
- callback = lambda namespace: 'custom handler'
- self.parser.register_subcommand(
- 'bar', self.get_sub_command(), callback=callback)
- namespace = self.parser.parse_args(['bar'])
- self.assertEqual(callback(namespace), namespace.main(namespace))
-
- def test_get_args_from_namespace(self):
+
+class BaseSubCommandTest(SubCommandFactoryTestMixin, unittest.TestCase):
+
+ def setUp(self):
+ self.namespace = argparse.Namespace()
+
+ def _successful_handler(self, namespace):
+ namespace.bar = True
+
+ def _failing_handler(self, namespace):
+ raise ValidationError('nothing is going on')
+
+ def test_needs_root(self):
+ # The subcommand's *needs_root* method, by default,
+ # returns *subcommand.root_required*.
+ subcommand = self.make_subcommand(root_required=True)
+ self.assertTrue(subcommand.needs_root(self.namespace))
+
+ def test_get_description(self):
+ # The default subcommand description is its own help.
+ subcommand = self.make_subcommand(help='subcommand help')
+ description = subcommand.get_description(self.namespace)
+ self.assertEqual(subcommand.help, description)
+
+ def test_prepare_namespace(self):
+ # The handlers are executed when *subcommand.prepare_namespace*
+ # is invoked.
+ subcommand = self.make_subcommand(handlers=[self._successful_handler])
+ subcommand.prepare_namespace(self.namespace)
+ self.assertTrue(self.namespace.bar)
+
+ def test_failing_handler(self):
+ # Ensure a validation error is raised by *subcommand.prepare_namespace*
+ # if the namespace is not valid.
+ subcommand = self.make_subcommand(handlers=[self._failing_handler])
+ with self.assertRaises(ValidationError) as cm:
+ subcommand.prepare_namespace(self.namespace)
+ self.assertIn('nothing is going on', str(cm.exception))
+
+
+class GetArgsFromNamespaceTest(ParserTestMixin, unittest.TestCase):
+
+ def test_different_namespace(self):
# It is possible to recreate the argument list taking values from
# a different namespace.
self.parser.add_argument('--foo')
self.parser.add_argument('bar')
namespace = self.parser.parse_args('--foo eggs spam'.split())
namespace.foo = 'changed'
- args = self.parser.get_args_from_namespace(namespace)
+ args = argparser.get_args_from_namespace(self.parser, namespace)
self.assertSequenceEqual(['--foo', 'changed', 'spam'], args)
- def test_args_from_namespace_with_multiple_values(self):
- # Ensure *get_args_from_namespace* correcty handles options
+ def test_with_multiple_values(self):
+ # Ensure *get_args_from_namespace* correctly handles options
# accepting multiple values.
self.parser.add_argument('foo')
self.parser.add_argument('--bar', nargs='+')
namespace = self.parser.parse_args('foo --bar eggs spam'.split())
namespace.bar.append('another argument')
- args = self.parser.get_args_from_namespace(namespace)
+ args = argparser.get_args_from_namespace(self.parser, namespace)
expected = ['foo', '--bar', 'eggs', 'spam', 'another argument']
self.assertSequenceEqual(expected, args)
- def test_args_from_namespace_with_boolean_values(self):
- # Ensure *get_args_from_namespace* correcty handles options
+ def test_boolean_values(self):
+ # Ensure *get_args_from_namespace* correctly handles options
# accepting boolean values.
self.parser.add_argument('--foo', action='store_true')
self.parser.add_argument('--bar', action='store_false')
expected = ['--foo', '--bar']
namespace = self.parser.parse_args(expected)
- args = self.parser.get_args_from_namespace(namespace)
+ args = argparser.get_args_from_namespace(self.parser, namespace)
self.assertSequenceEqual(expected, args)
- def test_help_subcommand(self):
- # Ensure the help sub command is added if other commands exist.
- self.parser.register_subcommand('foo', self.get_sub_command())
- namespace = self.parser.parse_args(['help'])
- with self.assertRaises(SystemExit) as cm:
- with capture_output() as output:
- namespace.main(namespace)
- self.assertEqual(0, cm.exception.code)
- self.assertIn('usage:', output.getvalue())
-
- def test_missing_help_subcommand(self):
- # Ensure the help sub command is missing if no other commands exist.
- with self.assertRaises(SystemExit) as cm:
- with capture_error() as error:
- self.parser.parse_args(['help'])
- self.assertEqual(2, cm.exception.code)
- self.assertIn('unrecognized arguments: help', error.getvalue())
-
-
-class BaseSubCommandTest(SubCommandTestMixin, unittest.TestCase):
-
- def set_handlers(self, *args):
- self.sub_command.handlers = args
-
- def _successful_handler(self, namespace):
- namespace.bar = True
-
- def _failing_handler(self, namespace):
- raise ValidationError('nothing is going on')
-
- def test_name(self):
- # Ensure a registered sub command has a name.
- self.assertEqual(self.sub_command_name, self.sub_command.name)
-
- def test_arguments(self):
- # Ensure the sub command arguments are correctly handled.
- namespace = self.parse_and_call_main('--foo', 'eggs')
- self.assertEqual('eggs', namespace.foo)
-
- def test_successful_validation(self):
- # Ensure attached handlers are called by the default callback.
- self.set_handlers(self._successful_handler)
- namespace = self.parse_and_call_main()
- self.assertTrue(namespace.bar)
-
- def test_failing_validation(self):
- # Ensure `ValidationError` stops the command execution.
- self.set_handlers(self._failing_handler)
- with self.assertRaises(SystemExit) as cm:
- with capture_error() as error:
- self.parse_and_call_main()
- self.assertEqual(2, cm.exception.code)
- self.assertIn('nothing is going on', error.getvalue())
-
- def test_help(self):
- # The help attribute of sub command instances is used to generate
- # the command usage message.
- help = self.parser.format_help()
- self.assertIn(self.sub_command.name, help)
- self.assertIn(self.sub_command.help, help)
-
- def test_init_namespace(self):
+
+class GetStepNameTest(StepsFactoryTestMixin, unittest.TestCase):
+
+ def test_named_step(self):
+ # Ensure the *step_name* attribute of a step, if present, is returned.
+ step = self.make_named_step('mystep')
+ self.assertEqual('mystep', argparser.get_step_name(step))
+
+ def test_function(self):
+ # If the *step_name* attribute is not present, *__name__* is used.
+ step = self.make_step()
+ self.assertEqual(step.__name__, argparser.get_step_name(step))
+
+
+class GetStepDescriptionTest(StepsFactoryTestMixin, unittest.TestCase):
+
+ def assertStartsWith(self, prefix, content):
+ """Assert that content starts with prefix."""
+ self.assertTrue(
+ content.startswith(prefix),
+ '%r does not start with %r' % (content, prefix))
+
+ def test_with_context(self):
+ # Ensure the description is correctly retrieved and formatted.
+ step = self.make_described_step('This step will do $stuff.')
+ description = argparser.get_step_description(step, stuff='nothing')
+ self.assertEqual('* This step will do nothing.', description)
+
+ def test_without_context(self):
+ # The description can be still retrieved if no context is provided.
+ original = 'This step will do $stuff.'
+ step = self.make_described_step(original)
+ description = argparser.get_step_description(step)
+ self.assertEqual('* ' + original, description)
+
+ def test_without_placeholder(self):
+ # Ensure an error is raised if a placeholder is missing.
+ step = self.make_described_step('This step will do $stuff.')
+ with self.assertRaises(KeyError):
+ argparser.get_step_description(step, foo='bar')
+
+ def test_missing_description(self):
+ # Ensure an empty string is returned if the description is None.
+ step = self.make_described_step(None)
+ description = argparser.get_step_description(step)
+ self.assertEqual('', description)
+
+ def test_no_description(self):
+ # Ensure an AttrubuteError is raised if the description is not found.
+ with self.assertRaises(AttributeError):
+ argparser.get_step_description(self.make_step())
+
+ def test_dedent(self):
+ # Ensure the description is correctly dedented.
+ original = """
+ Hi there!
+ """
+ step = self.make_described_step(original)
+ description = argparser.get_step_description(step)
+ self.assertEqual('* Hi there!', description)
+
+ def test_empty_lines(self):
+ # Ensure empty lines in description are removed.
+ step = self.make_described_step('Hello.\n \nGoodbye.')
+ description = argparser.get_step_description(step)
+ self.assertEqual('* Hello.\n* Goodbye.', description)
+
+ def test_wrapping(self):
+ # Ensure the description is correctly wrapped.
+ width = get_terminal_width()
+ elements = itertools.cycle('Lorem ipsum dolor sit amet.')
+ original = ''.join(itertools.islice(elements, width + 1))
+ step = self.make_described_step(original)
+ description = argparser.get_step_description(step)
+ lines = description.splitlines()
+ self.assertEqual(2, len(lines))
+ first_line, second_line = lines
+ self.assertStartsWith('* ', first_line)
+ self.assertStartsWith(' ', second_line)
+
+
+class InitNamespaceTest(unittest.TestCase):
+
+ def test_user_info_in_namespace(self):
# The namespace is initialized with current user info.
- namespace = self.parse()
+ namespace = argparse.Namespace()
+ argparser.init_namespace(namespace)
self.assertIsInstance(namespace.euid, int)
self.assertIsInstance(namespace.run_as_root, bool)
-class StepsBasedSubCommandTest(SubCommandTestMixin, unittest.TestCase):
-
- sub_command_class = examples.StepsBasedSubCommand
-
- def test_steps(self):
- # Ensure steps are executed in the order they are provided.
- with capture_output() as output:
- self.parse_and_call_main('--foo', 'eggs', '--bar', 'spam')
- self.check_output(
- ['step1 received eggs', 'step2 received eggs and spam'],
- output)
-
- def test_steps_flag(self):
- # A special argument `-s` or `--steps` is automatically added to the
- # parser. It can be used to execute only one or a subset of steps.
- with capture_output() as output:
- self.parse_and_call_main('--foo', 'eggs', '-s', 'step1')
- self.check_output(['step1 received eggs'], output)
-
- def test_skip_steps_flag(self):
- # A special argument `--skip-steps` is automatically added to the
- # parser. It can be used to skip one or more steps.
- with capture_output() as output:
- self.parse_and_call_main('--foo', 'eggs', '--skip-steps', 'step1')
- self.check_output(['step2 received eggs and None'], output)
-
- def test_step_name(self):
- # Ensure the string representation of a step is correctly retrieved.
- method = self.sub_command._get_step_name
- self.assertEqual('step1', method(examples.step1))
- self.assertEqual('mystep', method(examples.step2))
-
-
-class StepsBasedSubCommandWithErrorsTest(
- SubCommandTestMixin, unittest.TestCase):
-
- sub_command_class = examples.StepsBasedSubCommandWithErrors
-
- def test_failing_step(self):
- # Ensure the steps execution is stopped if a step raises
- # `subprocess.CalledProcessError`.
- with nested(
- capture_output(),
- self.assertRaises(subprocess.CalledProcessError)):
- self.parse_and_call_main('--foo', 'eggs')
-
-
-class DynamicStepsBasedSubCommandTest(SubCommandTestMixin, unittest.TestCase):
-
- sub_command_class = examples.DynamicStepsBasedSubCommand
-
- def test_dynamic_dispatcher(self):
- # The test runner calls a function named 'call_[step name]' if it is
- # defined.
- with capture_output() as output:
- self.parse_and_call_main('--foo', 'eggs', '--bar', 'spam')
- expected = [
- 'running step1 with eggs while bar is spam',
- 'step1 received eggs',
- 'step2 received eggs and spam'
+class ResolveStepsTest(StepsFactoryTestMixin, unittest.TestCase):
+
+ def setUp(self):
+ self.step1 = self.make_step('step1')
+ self.step2 = self.make_step('step2')
+ self.step3 = self.make_named_step('step3')
+ self.steps = [(self.step1, []), (self.step2, []), (self.step3, [])]
+
+ def resolve(self, steps, namespace):
+ """Resolve the steps and return them as an ordered dict."""
+ return OrderedDict(argparser.resolve_steps(steps, namespace))
+
+ def test_arguments_resolving(self):
+ # Ensure the steps arguments are correctly resolved using
+ # the current namespace.
+ steps = [(self.step1, ['arg1', 'arg2']), (self.step2, ['arg3'])]
+ namespace = argparse.Namespace(arg1='foo', arg2='bar', arg3='spam')
+ steps_dict = self.resolve(steps, namespace)
+ self.assertListEqual(['foo', 'bar'], steps_dict[self.step1])
+ self.assertListEqual(['spam'], steps_dict[self.step2])
+
+ def test_skip_steps(self):
+ # Ensure steps are correctly skipped if requested.
+ namespace = argparse.Namespace(skip_steps=['step1', 'step3'])
+ steps = self.resolve(self.steps, namespace).keys()
+ self.assertEqual(1, len(steps))
+ self.assertIs(self.step2, steps[0])
+
+ def test_include_steps(self):
+ # Ensure not explicitly included steps are excluded.
+ namespace = argparse.Namespace(steps=['step1', 'step3'])
+ steps = self.resolve(self.steps, namespace).keys()
+ self.assertListEqual([self.step1, self.step3], steps)
+
+ def test_include_skipped_steps(self):
+ # Ensure it's not possible to include a step already skipped.
+ namespace = argparse.Namespace(
+ steps=['step1'], skip_steps=['step1', 'step2', 'step3'])
+ self.assertEqual({}, self.resolve(self.steps, namespace))
+
+ def test_filter_function(self):
+ # The filter function can grant or deny the step execution.
+ steps = [
+ (self.step1, [], lambda namespace: namespace.run_step1),
+ (self.step2, [], lambda namespace: True)
]
- self.check_output(expected, output)
-
-
-class InteractiveStepsBasedSubCommandTest(
- SubCommandTestMixin, unittest.TestCase):
-
- sub_command_class = examples.InteractiveStepsBasedSubCommand
- step_description = examples.step_with_description.description
-
- def test_command_description(self):
- # Ensure the command description is generated collecting steps'
- # descriptions.
- with capture_output() as output:
- with RawInputReturning('yes'):
- self.parse_and_call_main()
- self.assertIn(self.step_description, output.getvalue())
-
- def test_interactive_execution_granted(self):
- # Ensure the command executes if the user confirms to proceed.
- with nested(capture_output(), RawInputReturning('yes')):
- retcode = self.parse_and_call_main()
- self.assertFalse(retcode)
-
- def test_interactive_execution_denied(self):
- # Ensure the command exits with an error if the user denies execution.
- with nested(capture_output(), RawInputReturning('no')):
- retcode = self.parse_and_call_main()
- self.assertEqual(1, retcode)
-
- def test_assume_yes(self):
- # Ensure confirmation is not asked if `--yes` is provided.
- with capture_output():
- with RawInputReturning('') as cm:
- self.parse_and_call_main('--yes')
- self.assertEqual(0, cm.call_count)
-
- def test_dry_run(self):
- # Ensure a dry run is never interactive, exits without errors and
- # prints out the command description.
- with capture_output() as output:
- with RawInputReturning('') as cm:
- retcode = self.parse_and_call_main('--dry-run')
- # Confirm has not been called.
- self.assertEqual(0, cm.call_count)
- # The command exits without errors.
- self.assertFalse(retcode)
- # The command description is displayed.
- self.assertIn(self.step_description, output.getvalue())
+ namespace = argparse.Namespace(run_step1=False)
+ steps = self.resolve(steps, namespace).keys()
+ self.assertEqual(1, len(steps))
+ self.assertIs(self.step2, steps[0])
+
+
+class RestartAsRootTest(ParserTestMixin, unittest.TestCase):
+
+ def test_restart_normal(self):
+ # Ensure an arbitrary command can be correctly restarted using `sudo`.
+ args = ['--foo', 'bar']
+ self.parser.add_argument(args[0])
+ namespace = self.parser.parse_args(args)
+ with call_replaced_by(lambda cmd: cmd):
+ cmd = argparser.restart_as_root(self.parser, namespace)
+ expected = ['sudo', sys.argv[0]] + args
+ self.assertListEqual(expected, cmd)
+
+ def test_restart_sub_command(self):
+ # Ensure a subcommand is correctly restarted using `sudo`.
+ name = 'subcommand-name'
+ subparsers = self.parser.add_subparsers()
+ subparser = subparsers.add_parser(name)
+ namespace = self.parser.parse_args([name])
+ with call_replaced_by(lambda cmd: cmd):
+ cmd = argparser.restart_as_root(subparser, namespace)
+ expected = ['sudo', sys.argv[0], name]
+ self.assertListEqual(expected, cmd)
+
+
+class StepsBasedSubCommandTest(
+ SubCommandFactoryTestMixin, StepsFactoryTestMixin, unittest.TestCase):
+
+ def make_subcommand_with_steps(self, *args):
+ """Create a steps based subcommand containing given steps.
+
+ Each step takes no arguments.
+ """
+ steps = [(step, []) for step in args]
+ return self.make_steps_based_subcommand(steps=steps)
+
+ def test_add_arguments(self):
+ # The special arguments `--steps` and `--skip-steps` are
+ # automatically added to the parser.
+ step1 = self.make_named_step('step1')
+ step2 = self.make_named_step('step2')
+ step3 = self.make_named_step('step3')
+ subcommand = self.make_subcommand_with_steps(step1, step2, step3)
+ parser = argparse.ArgumentParser()
+ subcommand.add_arguments(parser)
+ steps_to_skip = ['step1', 'step2']
+ steps_to_run = ['step3']
+ namespace = parser.parse_args(
+ ['--skip-steps'] + steps_to_skip + ['--steps'] + steps_to_run)
+ self.assertListEqual(steps_to_skip, namespace.skip_steps)
+ self.assertListEqual(steps_to_run, namespace.steps)
+
+ def test_get_description(self):
+ # Ensure the subcommand's description includes the description
+ # of each step.
+ step1 = self.make_described_step('step1 description')
+ step2 = self.make_described_step(None)
+ step3 = self.make_described_step('step3 description')
+ subcommand = self.make_subcommand_with_steps(step1, step2, step3)
+ description = subcommand.get_description(argparse.Namespace())
+ # A step with no description is ignored.
+ step1_description, step3_description = description.split('\n')
+ self.assertIn(argparser.get_step_description(step1), step1_description)
+ self.assertIn(argparser.get_step_description(step3), step3_description)
+
+ def test_run(self):
+ # Ensure the subcommand *run* method executes the steps
+ # in the order they are provided.
+ executed = []
+ # Each step adds its name to *executed* when is run.
+ step1 = lambda: executed.append('step1')
+ step2 = lambda: executed.append('step2')
+ step3 = lambda: executed.append('step3')
+ subcommand = self.make_subcommand_with_steps(step1, step2, step3)
+ subcommand.run(argparse.Namespace())
+ self.assertListEqual(['step1', 'step2', 'step3'], executed)
=== added file 'lpsetup/tests/test_cli.py'
--- lpsetup/tests/test_cli.py 1970-01-01 00:00:00 +0000
+++ lpsetup/tests/test_cli.py 2012-08-09 15:36:20 +0000
@@ -0,0 +1,299 @@
+#!/usr/bin/env python
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the cli module."""
+
+from contextlib import nested
+import unittest
+
+from lpsetup import (
+ cli,
+ exceptions,
+ )
+from lpsetup.tests.utils import (
+ call_replaced_by,
+ capture_error,
+ capture_output,
+ ParserTestMixin,
+ RawInputReturning,
+ SubCommandFactoryTestMixin,
+ )
+
+
+class GetParserTest(SubCommandFactoryTestMixin, unittest.TestCase):
+
+ def test_base_help(self):
+ # Ensure the help subcommand is correctly added to the parser.
+ parser = cli.get_parser([])
+ namespace = parser.parse_args(['help'])
+ with self.assertRaises(SystemExit) as cm:
+ with capture_output() as output:
+ namespace.main(namespace)
+ self.assertEqual(0, cm.exception.code)
+ self.assertIn('usage:', output.getvalue())
+
+ def test_subcommand_in_global_help(self):
+ # The help attribute of sub commands is used to generate
+ # the base command usage message.
+ subcommand = self.make_subcommand(help='help message')
+ parser = cli.get_parser([('foo', subcommand)])
+ help = parser.format_help()
+ self.assertIn(subcommand.help, help)
+ self.assertIn('foo', help)
+
+ def test_help_subcommand(self):
+ # Ensure the help sub command correctly show help messages retrieving
+ # info from other sub commands.
+ name = 'foo'
+ subcommand = self.make_subcommand(
+ help='subcommand help message',
+ add_arguments=lambda self, parser: parser.add_argument('--foo'))
+ parser = cli.get_parser([(name, subcommand)])
+ namespace = parser.parse_args(['help', name])
+ with self.assertRaises(SystemExit) as cm:
+ with capture_output() as output:
+ namespace.main(namespace)
+ output_value = output.getvalue()
+ # The help sub command exits without errors.
+ self.assertEqual(0, cm.exception.code)
+ # The sub command help is present in the output.
+ self.assertIn(subcommand.help, output_value)
+ # The sub command name is displayed as well.
+ self.assertIn(name, output_value)
+ # The help sub command also prints arguments info.
+ self.assertIn('--foo', output_value)
+
+ def test_subcommands(self):
+ # Ensure sub parsers are added for each registered sub command.
+ subcommands = (
+ ('foo', self.make_subcommand_returning('foo')),
+ ('bar', self.make_subcommand_returning('bar')),
+ )
+ parser = cli.get_parser(subcommands)
+ for name in dict(subcommands):
+ namespace = parser.parse_args([name])
+ self.assertEqual(name, namespace.main(namespace))
+
+
+class PrepareParserTestMixin(ParserTestMixin, SubCommandFactoryTestMixin):
+
+ def prepare_and_parse(self, subcommand, args=()):
+ """Prepare and parse *self.parser* passing given *subcommand*.
+
+ Return the resulting namespace.
+ """
+ cli.prepare_parser(self.parser, subcommand)
+ return self.parser.parse_args(args)
+
+
+class PrepareParserTest(PrepareParserTestMixin, unittest.TestCase):
+
+ def test_main_function_stored(self):
+ # Ensure *prepare_parser* stores the main function in the parser
+ # defaults, and that the main function calls *subcommand.run*.
+ subcommand = self.make_subcommand_returning('foo')
+ namespace = self.prepare_and_parse(subcommand)
+ self.assertEqual('foo', namespace.main(namespace))
+
+ def test_common_arguments(self):
+ # Ensure common arguments are added to the parser and to the resulting
+ # namespace.
+ subcommand = self.make_subcommand(
+ has_dry_run=True, has_interactive_run=True)
+ args = ('--dry-run', '--yes')
+ namespace = self.prepare_and_parse(subcommand, args)
+ self.assertTrue(namespace.dry_run)
+ self.assertFalse(namespace.interactive)
+
+ def test_subcommand_arguments(self):
+ # Ensure subcommand specific arguments are added to the parser as well.
+ subcommand = self.make_subcommand(
+ add_arguments=lambda self, parser: parser.add_argument('--foo'))
+ args = ('--foo', 'bar')
+ namespace = self.prepare_and_parse(subcommand, args)
+ self.assertEqual('bar', namespace.foo)
+
+
+class RunTest(PrepareParserTestMixin, unittest.TestCase):
+
+ def assertHasAttr(self, obj, attr):
+ """Assert given *obj* has *attr*."""
+ self.assertTrue(
+ hasattr(obj, attr),
+ '%r does not have %r attribute' % (obj, attr))
+
+ def make_described_subcommand(self, **kwargs):
+ """Create a sub command having a *get_description* method.
+
+ The created subcommand returns None.
+ """
+ get_description = lambda self, namespace: 'subcommand description'
+ return self.make_subcommand_returning(
+ None, get_description=get_description, **kwargs)
+
+ def make_subcommand_requiring_root(self, **kwargs):
+ """Create a sub command requiring root."""
+ needs_root = lambda self, namespace: True
+ return self.make_described_subcommand(needs_root=needs_root, **kwargs)
+
+ def test_init_namespace(self):
+ # Ensure *argparser.init_namespace* is invoked by *run*.
+ subcommand = self.make_subcommand_returning('foo')
+ namespace = self.prepare_and_parse(subcommand)
+ cli.run(self.parser, subcommand, namespace)
+ self.assertHasAttr(namespace, 'euid')
+ self.assertHasAttr(namespace, 'run_as_root')
+
+ def test_prepare_namespace(self):
+ # Ensure *subcommand.prepare_namespace*, if defined, is invoked.
+ def prepare_namespace(self, namespace):
+ namespace.prepared = True
+
+ subcommand = self.make_subcommand_returning(
+ 'foo', prepare_namespace=prepare_namespace)
+ namespace = self.prepare_and_parse(subcommand)
+ cli.run(self.parser, subcommand, namespace)
+ self.assertTrue(namespace.prepared)
+
+ def test_validation_error_in_prepare_namespace(self):
+ # Ensure the execution is stopped if *prepare_namespace* raises a
+ # ValidationError.
+ msg = 'Nothing is really going on.'
+
+ def prepare_namespace(self, namespace):
+ raise exceptions.ValidationError(msg)
+
+ subcommand = self.make_subcommand(prepare_namespace=prepare_namespace)
+ namespace = self.prepare_and_parse(subcommand)
+ with self.assertRaises(SystemExit) as cm:
+ with capture_error() as error:
+ cli.run(self.parser, subcommand, namespace)
+ # The program exits with an error.
+ self.assertEqual(2, cm.exception.code)
+ # The validation error is printed to stderr.
+ self.assertIn(msg, error.getvalue())
+
+ def test_dry_run(self):
+ # The subcommand's description is printed to stdout and the execution
+ # is stopped without errors if this is a dry run.
+ subcommand = self.make_described_subcommand(has_dry_run=True)
+ namespace = self.prepare_and_parse(subcommand, ['--dry-run'])
+ with capture_output() as output:
+ retcode = cli.run(self.parser, subcommand, namespace)
+ # The command description is displayed.
+ self.assertIn(subcommand.get_description(namespace), output.getvalue())
+ # The command exited without errors.
+ self.assertIsNone(retcode)
+
+ def test_dry_run_is_never_interactive(self):
+ # Ensure a dry run is never interactive.
+ subcommand = self.make_described_subcommand(
+ has_dry_run=True, has_interactive_run=True)
+ namespace = self.prepare_and_parse(subcommand, ['--dry-run'])
+ with nested(capture_output(), RawInputReturning('')) as cms:
+ cli.run(self.parser, subcommand, namespace)
+ # The confirm function has not been called.
+ self.assertEqual(0, cms[1].call_count)
+
+ def test_dry_run_never_requires_root(self):
+ # Ensure a dry run never requires to restart as root.
+ subcommand = self.make_subcommand_requiring_root(has_dry_run=True)
+ namespace = self.prepare_and_parse(subcommand, ['--dry-run'])
+ with capture_output():
+ retcode = cli.run(self.parser, subcommand, namespace)
+ # The command exited without errors and without a restart as root.
+ self.assertIsNone(retcode)
+
+ def test_interactive_execution_granted(self):
+ # Ensure the execution continues if the user confirms to proceed.
+ subcommand = self.make_subcommand_returning(
+ None, has_interactive_run=True)
+ namespace = self.prepare_and_parse(subcommand)
+ with nested(capture_output(), RawInputReturning('yes')) as cms:
+ retcode = cli.run(self.parser, subcommand, namespace)
+ # The command exited without errors.
+ self.assertIsNone(retcode)
+ # The user confirmed the execution.
+ self.assertEqual(1, cms[1].call_count)
+
+ def test_interactive_execution_denied(self):
+ # Ensure the command exits with an error if the user denies execution.
+ subcommand = self.make_subcommand_returning(
+ None, has_interactive_run=True)
+ namespace = self.prepare_and_parse(subcommand)
+ with nested(capture_output(), RawInputReturning('no')):
+ retcode = cli.run(self.parser, subcommand, namespace)
+ self.assertEqual(1, retcode)
+
+ def test_assume_yes(self):
+ # Ensure confirmation is not asked if `--yes` is provided.
+ subcommand = self.make_subcommand_returning(
+ None, has_interactive_run=True)
+ namespace = self.prepare_and_parse(subcommand, ['--yes'])
+ with nested(capture_output(), RawInputReturning('')) as cms:
+ retcode = cli.run(self.parser, subcommand, namespace)
+ # The confirm function has not been called.
+ self.assertEqual(0, cms[1].call_count)
+ # The command exited without errors.
+ self.assertIsNone(retcode)
+
+ def test_restart_as_root(self):
+ # Ensure the execution is restarted as root if required.
+ subcommand = self.make_subcommand_requiring_root()
+ namespace = self.prepare_and_parse(subcommand)
+ with call_replaced_by(lambda cmd: cmd):
+ cmd = cli.run(self.parser, subcommand, namespace)
+ self.assertEqual('sudo', cmd[0])
+
+ def test_execution_error_in_run(self):
+ # An error is returned if the subcommand's *run()*
+ # raises an ExecutionError
+ execution_error = exceptions.ExecutionError()
+
+ def run(self, namespace):
+ raise execution_error
+
+ subcommand = self.make_subcommand(run=run)
+ namespace = self.prepare_and_parse(subcommand)
+ error = cli.run(self.parser, subcommand, namespace)
+ self.assertIs(execution_error, error)
+
+
+class MainTest(PrepareParserTestMixin, unittest.TestCase):
+
+ def test_normal_execution(self):
+ # Ensure *main* executes the main function as it is found
+ # in the namespace, and return its return value.
+ self.parser.set_defaults(main=lambda namespace: 1)
+ self.assertEqual(1, cli.main([], self.parser))
+
+ def test_subcommand_execution(self):
+ # Ensure *main* returns the subcommand's *run()* return value.
+ subcommand = self.make_subcommand_returning(1)
+ parser = cli.get_parser([('foo', subcommand)])
+ self.assertEqual(1, cli.main(['foo'], parser))
+
+ def test_help(self):
+ # Ensure *main* returns a non-error exit code if help is invoked.
+ with capture_output():
+ self.assertEqual(0, cli.main(['-h'], self.parser))
+
+ def test_subcommand_help(self):
+ # Ensure *main* returns a non-error exit code if a subcommand's help
+ # is requested.
+ subcommand = self.make_subcommand(help='help message')
+ parser = cli.get_parser([('foo', subcommand)])
+ with capture_output() as output:
+ retcode = cli.main(['help', 'foo'], parser)
+ self.assertEqual(0, retcode)
+ self.assertIn(subcommand.help, output.getvalue())
+
+ def test_invalid_arguments(self):
+ # Ensure *main* returns an error exit code if wrong arguments are
+ # provided.
+ invalid_args = 'invalid arguments'
+ with capture_error() as error:
+ retcode = cli.main(invalid_args.split(), self.parser)
+ self.assertEqual(2, retcode)
+ self.assertIn(invalid_args, error.getvalue())
=== modified file 'lpsetup/tests/test_utils.py'
--- lpsetup/tests/test_utils.py 2012-08-09 14:32:10 +0000
+++ lpsetup/tests/test_utils.py 2012-08-09 15:36:20 +0000
@@ -6,7 +6,6 @@
from datetime import datetime
import getpass
-import itertools
import os
import shutil
import sys
@@ -32,7 +31,6 @@
get_lxc_gateway,
get_network_interfaces,
get_running_containers,
- get_step_description,
get_terminal_width,
render_to_file,
retry,
@@ -207,74 +205,6 @@
self.assertRunning([], ['c1', 'c2', 'c3'])
-class GetStepDescriptionTest(unittest.TestCase):
-
- def assertStartsWith(self, prefix, content):
- """Assert that content starts with prefix."""
- self.assertTrue(
- content.startswith(prefix),
- '%r does not start with %r' % (content, prefix))
-
- def get_step(self, description=None):
- step = lambda: None
- step.description = description
- return step
-
- def test_with_context(self):
- # Ensure the description is correctly retrieved and formatted.
- step = self.get_step('This step will do $stuff.')
- description = get_step_description(step, stuff='nothing')
- self.assertEqual('* This step will do nothing.', description)
-
- def test_without_context(self):
- # The description can be still retrieved if no context is provided.
- original = 'This step will do $stuff.'
- description = get_step_description(self.get_step(original))
- self.assertEqual('* ' + original, description)
-
- def test_without_placeholder(self):
- # Ensure an error is raised if a placeholder is missing.
- expected = 'This step will do $stuff.'
- with self.assertRaises(KeyError):
- get_step_description(self.get_step(expected), foo='bar')
-
- def test_missing_description(self):
- # Ensure an empty string is returned if the description is None.
- description = get_step_description(self.get_step())
- self.assertEqual('', description)
-
- def test_no_description(self):
- # Ensure an AttrubuteError is raised if the description is not found.
- with self.assertRaises(AttributeError):
- get_step_description(lambda: None)
-
- def test_dedent(self):
- # Ensure the description is correctly dedented.
- original = """
- Hi there!
- """
- description = get_step_description(self.get_step(original))
- self.assertEqual('* Hi there!', description)
-
- def test_empty_lines(self):
- # Ensure empty lines in description are removed.
- original = 'Hello.\n \nGoodbye.'
- description = get_step_description(self.get_step(original))
- self.assertEqual('* Hello.\n* Goodbye.', description)
-
- def test_wrapping(self):
- # Ensure the description is correctly wrapped.
- width = get_terminal_width()
- elements = itertools.cycle('Lorem ipsum dolor sit amet.')
- original = ''.join(itertools.islice(elements, width + 1))
- description = get_step_description(self.get_step(original))
- lines = description.splitlines()
- self.assertEqual(2, len(lines))
- first_line, second_line = lines
- self.assertStartsWith('* ', first_line)
- self.assertStartsWith(' ', second_line)
-
-
class GetTerminalSizeTest(unittest.TestCase):
def test_is_integer(self):
=== modified file 'lpsetup/tests/utils.py'
--- lpsetup/tests/utils.py 2012-07-30 13:37:25 +0000
+++ lpsetup/tests/utils.py 2012-08-09 15:36:20 +0000
@@ -6,11 +6,17 @@
__metaclass__ = type
__all__ = [
+ 'call_replaced_by',
+ 'capture',
'capture_error',
'capture_output',
+ 'create_test_branch',
'get_random_string',
- 'StepsBasedSubCommandTestMixin',
- 'SubCommandTestMixin',
+ 'ParserTestMixin',
+ 'RawInputReturning',
+ 'skip_if_no_lpuser',
+ 'StepsFactoryTestMixin',
+ 'SubCommandFactoryTestMixin',
]
from contextlib import contextmanager
@@ -30,12 +36,40 @@
run,
)
-from lpsetup import argparser
-from lpsetup.tests import examples
+from lpsetup import (
+ argparser,
+ cli,
+ )
from lpsetup.utils import call
@contextmanager
+def call_replaced_by(function):
+ """Temporarily mock *subprocess.call* with the given *function*."""
+ original, subprocess.call = subprocess.call, function
+ try:
+ yield
+ finally:
+ subprocess.call = original
+
+
+class CallReplacedByTest(unittest.TestCase):
+ """Tests for the *call_replaced_by* context manager."""
+
+ def test_call_is_replaced(self):
+ # Ensure *subprocess.call* is correctly replaced.
+ with call_replaced_by(lambda *args, **kwargs: 'mocked'):
+ self.assertEqual('mocked', subprocess.call(['ls']))
+
+ def test_call_is_restored(self):
+ # Ensure *subprocess.call* is correctly restored exiting from the cm.
+ original = subprocess.call
+ with call_replaced_by(lambda *args, **kwargs: 'mocked'):
+ pass
+ self.assertIs(original, subprocess.call)
+
+
+@contextmanager
def capture(attr):
output = StringIO()
backup = getattr(sys, attr)
@@ -124,8 +158,15 @@
lpuser is None, 'You need to set up a Launchpad login to run this test.')
+class ParserTestMixin(object):
+
+ def setUp(self):
+ """Set up an argument parser."""
+ self.parser = argparser.ArgumentParser()
+
+
class RawInputReturning(object):
- """Mocks the *raw_input* builtin function.
+ """Mocks the *raw_input* built-in function.
This context manager takes one or more pre-defined answers and
keeps track of mocked *raw_input* call count.
@@ -143,70 +184,96 @@
def __exit__(self, exc_type, exc_val, exc_tb):
self._builtin.raw_input = self.original
- def input(self, question):
+ def input(self, prompt=''):
self.call_count += 1
return self.answers.next()
-@contextmanager
-def root_not_needed(subcommand):
- """Temporarily set to False the *needs_root* flag of *subcommand*."""
- original, subcommand.needs_root = subcommand.needs_root, False
- try:
- yield
- finally:
- subcommand.needs_root = original
-
-
-class RootNotNeededTest(unittest.TestCase):
-
- def test_context_manager(self):
- # Ensure the context manager temporarily sets to False *needs_root*.
- subcommand = type(
- 'SubCommand', (argparser.BaseSubCommand,), {'needs_root': True})
- with root_not_needed(subcommand):
- self.assertFalse(subcommand.needs_root)
- self.assertTrue(subcommand.needs_root)
-
-
-class SubCommandTestMixin(object):
-
- sub_command_class = examples.SubCommand
- sub_command_name = 'subcmd'
-
- def setUp(self):
- """Set up an argument parser and instantiate *self.sub_command_class*.
-
- The name used to create the sub command instance is
- *self.sub_command_name*.
- """
- self.parser = argparser.ArgumentParser()
- self.sub_command = self.parser.register_subcommand(
- self.sub_command_name, self.sub_command_class)
-
- def parse(self, *args):
- """Parse given *args* and return an initialized namespace object."""
- namespace = self.parser.parse_args((self.sub_command_name,) + args)
- sub_command = self.sub_command
- sub_command.init_namespace(namespace)
- sub_command.prepare_namespace(self.parser, namespace)
- return namespace
-
- def parse_and_call_main(self, *args):
- """Create a namespace using the given *args* and invoke main."""
- namespace = self.parse(*args)
- return namespace.main(namespace)
-
- def check_output(self, expected, output):
- value = filter(None, output.getvalue().split('\n'))
- self.assertSequenceEqual(expected, value)
-
-
-class StepsBasedSubCommandTestMixin(SubCommandTestMixin):
+class RawInputReturningTest(unittest.TestCase):
+ """Tests for the *RawInputReturning* context manager."""
+
+ def test_original_is_replaced(self):
+ # Ensure the original function is correctly replaced.
+ with RawInputReturning('mocked'):
+ self.assertEqual('mocked', raw_input())
+
+ def test_original_is_retrievable(self):
+ # Ensure the original function is still stored as an attribute of the
+ # context manager.
+ original = raw_input
+ with RawInputReturning('mocked') as cm:
+ self.assertIs(original, cm.original)
+
+ def test_multiple_return_values(self):
+ # Each call returns one of the provided values in order.
+ with RawInputReturning(*range(2)):
+ self.assertEqual(0, raw_input())
+ self.assertEqual(1, raw_input())
+
+ def test_call_count(self):
+ # Ensure the context manager keeps track of call count.
+ return_values = range(2)
+ with RawInputReturning(*return_values) as cm:
+ self.assertEqual(0, cm.call_count)
+ for i in return_values:
+ raw_input()
+ self.assertEqual(2, cm.call_count)
+
+
+class StepsFactoryTestMixin(object):
+
+ def make_step(self, name='step'):
+ """Return a step callable."""
+ def step(*args):
+ return args
+
+ step.__name__ = name
+ return step
+
+ def make_named_step(self, step_name):
+ """Return a named step callable."""
+ step = self.make_step()
+ step.step_name = step_name
+ return step
+
+ def make_described_step(self, description):
+ """Return a step having a description."""
+ step = self.make_step()
+ step.description = description
+ return step
+
+
+class SubCommandFactoryTestMixin(object):
+
+ def make_subcommand(self, **kwargs):
+ """Create and return a sub command.
+
+ Attributes can be specified using kwargs.
+ """
+ return type('SubCommand', (argparser.BaseSubCommand,), kwargs)()
+
+ def make_subcommand_returning(self, value, **kwargs):
+ """Create a sub command having the run function returning *value*."""
+ run = lambda self, namespace: value
+ return self.make_subcommand(run=run, **kwargs)
+
+ def make_steps_based_subcommand(self, **kwargs):
+ """Create and return a steps based sub command.
+
+ Attributes can be specified using kwargs.
+ """
+ return type(
+ 'StepsBasedSubCommand',
+ (argparser.StepsBasedSubCommand,),
+ kwargs)()
+
+
+class StepsBasedSubCommandTestMixin(ParserTestMixin):
"""This mixin can be used to test sub commands steps and handlers.
Real TestCases subclassing this mixin must define:
+ - sub_command_class: the steps based subcommand class
- expected_arguments: a sequence of command line arguments
used by the current tested sub command
- expected_handlers: a sequence of expected handler callables
@@ -219,17 +286,21 @@
def setUp(self):
"""Set up a namespace using *self.expected_arguments*."""
super(StepsBasedSubCommandTestMixin, self).setUp()
- self.namespace = self.parse(*self.expected_arguments)
+ self.subcommand = self.sub_command_class()
+ cli.prepare_parser(self.parser, self.subcommand)
+ self.namespace = self.parser.parse_args(self.expected_arguments)
+ argparser.init_namespace(self.namespace)
def test_handlers(self):
# Ensure this sub command uses the expected handlers.
- handlers = self.sub_command.get_handlers(self.namespace)
+ handlers = self.subcommand.get_handlers(self.namespace)
self.assertSequenceEqual(self.expected_handlers, handlers)
def test_steps(self):
# Ensure this sub command wants to run the expected steps.
- steps = self.sub_command.get_steps(self.namespace)
- real_steps = [[step, list(args)] for _, step, args in steps]
+ self.subcommand.prepare_namespace(self.namespace)
+ steps = argparser.resolve_steps(self.subcommand.steps, self.namespace)
+ real_steps = [[step, list(args)] for step, args in steps]
expected_steps = []
for step, arg_names in self.expected_steps:
args = [getattr(self.namespace, name) for name in arg_names]
@@ -238,5 +309,5 @@
def test_needs_root(self):
# The root user may or may not be required to run this sub command.
- needs_root = self.sub_command.get_needs_root(self.namespace)
+ needs_root = self.subcommand.needs_root(self.namespace)
self.assertEqual(self.needs_root, needs_root)
=== modified file 'lpsetup/utils.py'
--- lpsetup/utils.py 2012-08-09 14:33:33 +0000
+++ lpsetup/utils.py 2012-08-09 15:36:20 +0000
@@ -13,7 +13,6 @@
'get_lxc_gateway',
'get_network_interfaces',
'get_running_containers',
- 'get_step_description',
'get_terminal_width',
'lxc_in_state',
'lxc_ip',
@@ -39,7 +38,6 @@
import re
import subprocess
import shutil
-import string
import sys
import textwrap
import time
@@ -216,34 +214,6 @@
visited[container] = 1
-def get_step_description(step, **kwargs):
- """Retrieve and format step description from the given *step* callable.
-
- *kwargs*, if provided, will be used as context to format the description.
- Formatting is done using the Python's builtin templating system
- *string.Template* supporting $-based substitutions.
-
- If placeholders are missing from *kwargs* an error will be raised.
- """
- description = step.description
- if not description:
- # This step has no description, nothing else to do.
- return ''
- if kwargs:
- s = string.Template(description)
- description = s.substitute(**kwargs)
- # Remove multiple spaces from lines.
- lines = [' '.join(line.split()) for line in description.splitlines()]
- # Retrieve all the non empty lines.
- lines = filter(None, lines)
- # For each line, wrap the contents. Note that we can't wrap the text of
- # the entire paragraph because we want to preserve existing new lines.
- width = get_terminal_width()
- return '\n'.join(textwrap.fill(
- line, width=width, initial_indent='* ', subsequent_indent=' ')
- for line in lines)
-
-
def get_terminal_width():
"""Return the terminal width (number of columns)."""
try:
Follow ups
