andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request #570 from adamcik/feature/subcommands

Browse Source 
Add sub-commands to mopidy and switch mopidy-scan over.
This commit is contained in:
Stein Magnus Jodal 2013-11-16 05:00:42 -08:00
commit c97e812544
22 changed files with 1114 additions and 444 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
docs/api/audio.rst
View File

@ -28,3 +28,10 @@ Audio listener
.. autoclass:: mopidy.audio.AudioListener
:members:
Audio scanner
=============
.. autoclass:: mopidy.audio.scan.Scanner
:members:

9
docs/api/commands.rst Normal file
View File

@ -0,0 +1,9 @@
.. _commands-api:
************
Commands API
************
.. automodule:: mopidy.commands
:synopsis: Commands API for Mopidy CLI.
:members:

1
docs/api/index.rst
View File

@ -13,6 +13,7 @@ API reference
core
audio
frontends
commands
ext
config
http

15
docs/changelog.rst
View File

@ -74,6 +74,21 @@ v0.17.0 (UNRELEASED)
:confval:`http/zeroconf` config value to change the service name or disable
the service. (Fixes: :issue:`39`)
**Sub-commands**
- Switched to sub-commands for the ``mopidy`` command , this implies the
following changes (fixes :issue:`437`):
===================== =================
Old command New command
===================== =================
mopidy --show-deps mopidy deps
mopidy --show-config mopidy config
mopidy-scan mopidy local scan
- Added hooks for extensions to create their own custom sub-commands and
converted ``mopidy-scan`` as first user of new API. (Fixes :issue:`436`)
v0.16.1 (2013-11-02)
====================

59
docs/commands/mopidy-scan.rst
View File

@ -1,59 +0,0 @@
.. _mopidy-scan-cmd:
*******************
mopidy-scan command
*******************
Synopsis
========
mopidy-scan
[-h] [--version] [-q] [-v]
Description
===========
Mopidy is a music server which can play music both from multiple sources, like
your local hard drive, radio streams, and from Spotify and SoundCloud. Searches
combines results from all music sources, and you can mix tracks from all
sources in your play queue. Your playlists from Spotify or SoundCloud are also
available for use.
The ``mopidy-scan`` command is used to index a music library to make it
available for playback with ``mopidy``.
Options
=======
.. program:: mopidy-scan
.. cmdoption:: --version
Show Mopidy's version number and exit.
.. cmdoption:: -h, --help
Show help message and exit.
.. cmdoption:: -q, --quiet
Show less output: warning level and higher.
.. cmdoption:: -v, --verbose
Show more output: debug level and higher.
See also
========
:ref:`mopidy(1) <mopidy-cmd>`
Reporting bugs
==============
Report bugs to Mopidy's issue tracker at
<https://github.com/mopidy/mopidy/issues>

50
docs/commands/mopidy.rst
View File

@ -8,8 +8,8 @@ Synopsis
========
mopidy
[-h] [--version] [-q] [-v] [--save-debug-log] [--show-config]
[--show-deps] [--config CONFIG_FILES] [-o CONFIG_OVERRIDES]
[-h] [--version] [-q] [-v] [--save-debug-log] [--config CONFIG_FILES]
[-o CONFIG_OVERRIDES] [COMMAND] ...
Description
@ -50,20 +50,10 @@ Options
Save debug log to the file specified in the :confval:`logging/debug_file`
config value, typically ``./mopidy.log``.
.. cmdoption:: --show-config
Show the current effective config. All configuration sources are merged
together to show the effective document. Secret values like passwords are
masked out. Config for disabled extensions are not included.
.. cmdoption:: --show-deps
Show dependencies, their versions and installation location.
.. cmdoption:: --config <file>
Specify config file to use. To use multiple config files, separate them
with colon. The later files override the earlier ones if there's a
with a colon. The later files override the earlier ones if there's a
conflict.
.. cmdoption:: -o <option>, --option <option>
@ -72,6 +62,32 @@ Options
be provided multiple times.
Built in commands
=================
.. cmdoption:: config
Show the current effective config. All configuration sources are merged
together to show the effective document. Secret values like passwords are
masked out. Config for disabled extensions are not included.
.. cmdoption:: deps
Show dependencies, their versions and installation location.
Extension commands
==================
Additionally, extensions can provide extra commands. See ``mopidy --help``
for a list of what is available on your system and command-specific help.
Commands for disabled extensions will be listed, but can not be run.
.. cmdoption:: local scan
Scan local media files present in your library.
Files
=====
@ -105,17 +121,15 @@ configs::
mopidy -o mpd/enabled=false -o spotify/bitrate=320
The :option:`--show-config` output shows the effect of the :option:`--option`
flags::
The ``mopidy config`` output shows the effect of the :option:`--option` flags::
mopidy -o mpd/enabled=false -o spotify/bitrate=320 --show-config
mopidy -o mpd/enabled=false -o spotify/bitrate=320 config
See also
========
:ref:`mopidy-scan(1) <mopidy-scan-cmd>`, :ref:`mopidy-convert-config(1)
<mopidy-convert-config>`
:ref:`mopidy-convert-config(1) <mopidy-convert-config>`
Reporting bugs
==============

7
docs/conf.py
View File

@ -141,13 +141,6 @@ man_pages = [
'',
'1'
),
(
'commands/mopidy-scan',
'mopidy-scan',
'index music for playback with mopidy',
'',
'1'
),
(
'commands/mopidy-convert-config',
'mopidy-convert-config',

17
docs/ext/local.rst
View File

@ -6,7 +6,7 @@ Mopidy-Local
Extension for playing music from a local music archive.
This backend handles URIs starting with ``file:``.
This backend handles URIs starting with ``local:``.
Known issues
@ -71,7 +71,7 @@ music...
Generating a tag cache
----------------------
The program :command:`mopidy-scan` will scan the path set in the
The command :command:`mopidy local scan` will scan the path set in the
:confval:`local/media_dir` config value for any media files and build a MPD
compatible ``tag_cache``.
@ -80,16 +80,11 @@ To make a ``tag_cache`` of your local music available for Mopidy:
#. Ensure that the :confval:`local/media_dir` config value points to where your
music is located. Check the current setting by running::
mopidy --show-config
mopidy config
#. Scan your media library. The command outputs the ``tag_cache`` to
standard output, which means that you will need to redirect the output to a
file yourself::
#. Scan your media library. The command writes the ``tag_cache`` to
the :confval:`local/tag_cache_file`::
mopidy-scan > tag_cache
#. Move the ``tag_cache`` file to the location
set in the :confval:`local/tag_cache_file` config value, or change the
config value to point to where your ``tag_cache`` file is.
mopidy local scan
#. Start Mopidy, find the music library in a client, and play some local music!

35
docs/extensiondev.rst
View File

@ -305,6 +305,10 @@ This is ``mopidy_soundspot/__init__.py``::
from .backend import SoundspotBackend
return [SoundspotBackend]
def get_command(self):
from .commands import SoundspotCommand
return SoundspotCommand()
def register_gstreamer_elements(self):
from .mixer import SoundspotMixer
gobject.type_register(SoundspotMixer)
@ -353,7 +357,8 @@ Example backend
If you want to extend Mopidy to support new music and playlist sources, you
want to implement a backend. A backend does not have access to Mopidy's core
API at all and got a bunch of interfaces to implement.
API at all, but it does have a bunch of interfaces it can implement to extend
Mopidy.
The skeleton of a backend would look like this. See :ref:`backend-api` for more
details.
@ -373,6 +378,34 @@ details.
# Your backend implementation
Example command
===============
If you want to extend the Mopidy with a new helper not run from the server,
such as scanning for media, adding a command is the way to go. Your top level
command name will always match your extension name, but you are free to add
sub-commands with names of your choosing.
The skeleton of a commands would look like this. See :ref:`command-api` for more
details.
::
from mopidy import commands
class SoundspotCommand(commands.Command):
help = 'Some text that will show up in --help'
def __init__(self):
super(SoundspotCommand, self).__init__()
self.add_argument('--foo')
def run(self, args, config, extensions):
# Your backend implementation
return 0
Example GStreamer element
=========================

2
docs/running.rst
View File

@ -7,7 +7,7 @@ To start Mopidy, simply open a terminal and run::
mopidy
For a complete reference to the Mopidy commands and their command line options,
see :ref:`mopidy-cmd` and :ref:`mopidy-scan-cmd`.
see :ref:`mopidy-cmd`.
When Mopidy says ``MPD server running at [127.0.0.1]:6600`` it's ready to
accept connections by any MPD client. Check out our non-exhaustive

4
docs/troubleshooting.rst
View File

@ -28,7 +28,7 @@ accepted, but large logs should still be shared through a pastebin.
Effective configuration
=======================
The command :option:`mopidy --show-config` will print your full effective
The command ``mopidy config`` will print your full effective
configuration the way Mopidy sees it after all defaults and all config files
have been merged into a single config document. Any secret values like
passwords are masked out, so the output of the command should be safe to share
@ -38,7 +38,7 @@ with others for debugging.
Installed dependencies
======================
The command :option:`mopidy --show-deps` will list the paths to and versions of
The command ``mopidy deps`` will list the paths to and versions of
any dependency Mopidy or the extensions might need to work. This is very useful
data for checking that you're using the right versions, and that you're using
the right installation if you have multiple installations of a dependency on

180
mopidy/__main__.py
View File

@ -18,72 +18,100 @@ sys.argv[1:] = []
from mopidy import commands, ext
from mopidy.audio import Audio
from mopidy import config as config_lib
from mopidy.core import Core
from mopidy.utils import log, path, process
from mopidy.utils import log, path, process, versioning
logger = logging.getLogger('mopidy.main')
def main():
log.bootstrap_delayed_logging()
logger.info('Starting Mopidy %s', versioning.get_version())
signal.signal(signal.SIGTERM, process.exit_handler)
signal.signal(signal.SIGUSR1, pykka.debug.log_thread_tracebacks)
args = commands.parser.parse_args(args=mopidy_args)
if args.show_config:
commands.show_config(args)
if args.show_deps:
commands.show_deps()
# TODO: figure out a way to make the boilerplate in this file reusable in
# scanner and other places we need it.
try:
# Initial config without extensions to bootstrap logging.
logging_initialized = False
logging_config, _ = config_lib.load(
args.config_files, [], args.config_overrides)
# TODO: setup_logging needs defaults in-case config values are None
log.setup_logging(
logging_config, args.verbosity_level, args.save_debug_log)
logging_initialized = True
create_file_structures()
check_old_locations()
root_cmd = commands.RootCommand()
config_cmd = commands.ConfigCommand()
deps_cmd = commands.DepsCommand()
root_cmd.set(extension=None)
root_cmd.add_child('config', config_cmd)
root_cmd.add_child('deps', deps_cmd)
installed_extensions = ext.load_extensions()
for extension in installed_extensions:
ext_cmd = extension.get_command()
if ext_cmd:
ext_cmd.set(extension=extension)
root_cmd.add_child(extension.ext_name, ext_cmd)
args = root_cmd.parse(mopidy_args)
config, config_errors = config_lib.load(
args.config_files, installed_extensions, args.config_overrides)
# Filter out disabled extensions and remove any config errors for them.
verbosity_level = args.base_verbosity_level
if args.verbosity_level:
verbosity_level += args.verbosity_level
log.setup_logging(config, verbosity_level, args.save_debug_log)
enabled_extensions = []
for extension in installed_extensions:
enabled = config[extension.ext_name]['enabled']
if ext.validate_extension(extension) and enabled:
if not ext.validate_extension(extension):
config[extension.ext_name] = {'enabled': False}
config_errors[extension.ext_name] = {
'enabled': 'extension disabled by self check.'}
elif not config[extension.ext_name]['enabled']:
config[extension.ext_name] = {'enabled': False}
config_errors[extension.ext_name] = {
'enabled': 'extension disabled by user config.'}
else:
enabled_extensions.append(extension)
elif extension.ext_name in config_errors:
del config_errors[extension.ext_name]
log_extension_info(installed_extensions, enabled_extensions)
ext.register_gstreamer_elements(enabled_extensions)
# Config and deps commands are simply special cased for now.
if args.command == config_cmd:
return args.command.run(
config, config_errors, installed_extensions)
elif args.command == deps_cmd:
return args.command.run()
# Remove errors for extensions that are not enabled:
for extension in installed_extensions:
if extension not in enabled_extensions:
config_errors.pop(extension.ext_name, None)
check_config_errors(config_errors)
# Read-only config from here on, please.
proxied_config = config_lib.Proxy(config)
log.setup_log_levels(proxied_config)
ext.register_gstreamer_elements(enabled_extensions)
if args.extension and args.extension not in enabled_extensions:
logger.error(
'Unable to run command provided by disabled extension %s',
args.extension.ext_name)
return 1
# Anything that wants to exit after this point must use
# mopidy.utils.process.exit_process as actors have been started.
start(proxied_config, enabled_extensions)
# mopidy.utils.process.exit_process as actors can have been started.
try:
return args.command.run(args, proxied_config, enabled_extensions)
except NotImplementedError:
print root_cmd.format_help()
return 1
except KeyboardInterrupt:
pass
except Exception as ex:
if logging_initialized:
logger.exception(ex)
logger.exception(ex)
raise
@ -127,89 +155,5 @@ def check_config_errors(errors):
sys.exit(1)
def start(config, extensions):
loop = gobject.MainLoop()
try:
audio = start_audio(config)
backends = start_backends(config, extensions, audio)
core = start_core(audio, backends)
start_frontends(config, extensions, core)
loop.run()
except KeyboardInterrupt:
logger.info('Interrupted. Exiting...')
return
finally:
loop.quit()
stop_frontends(extensions)
stop_core()
stop_backends(extensions)
stop_audio()
process.stop_remaining_actors()
def start_audio(config):
logger.info('Starting Mopidy audio')
return Audio.start(config=config).proxy()
def stop_audio():
logger.info('Stopping Mopidy audio')
process.stop_actors_by_class(Audio)
def start_backends(config, extensions, audio):
backend_classes = []
for extension in extensions:
backend_classes.extend(extension.get_backend_classes())
logger.info(
'Starting Mopidy backends: %s',
', '.join(b.__name__ for b in backend_classes) or 'none')
backends = []
for backend_class in backend_classes:
backend = backend_class.start(config=config, audio=audio).proxy()
backends.append(backend)
return backends
def stop_backends(extensions):
logger.info('Stopping Mopidy backends')
for extension in extensions:
for backend_class in extension.get_backend_classes():
process.stop_actors_by_class(backend_class)
def start_core(audio, backends):
logger.info('Starting Mopidy core')
return Core.start(audio=audio, backends=backends).proxy()
def stop_core():
logger.info('Stopping Mopidy core')
process.stop_actors_by_class(Core)
def start_frontends(config, extensions, core):
frontend_classes = []
for extension in extensions:
frontend_classes.extend(extension.get_frontend_classes())
logger.info(
'Starting Mopidy frontends: %s',
', '.join(f.__name__ for f in frontend_classes) or 'none')
for frontend_class in frontend_classes:
frontend_class.start(config=config, core=core)
def stop_frontends(extensions):
logger.info('Stopping Mopidy frontends')
for extension in extensions:
for frontend_class in extension.get_frontend_classes():
process.stop_actors_by_class(frontend_class)
if __name__ == '__main__':
main()
sys.exit(main())

16
mopidy/audio/scan.py
View File

@ -14,6 +14,15 @@ from mopidy.utils import path
class Scanner(object):
"""
Helper to get tags and other relevant info from URIs.
:param timeout: timeout for scanning a URI in ms
:type event: int
:param min_duration: minimum duration of scanned URI in ms, -1 for all.
:type event: int
"""
def __init__(self, timeout=1000, min_duration=100):
self.timeout_ms = timeout
self.min_duration_ms = min_duration
@ -35,6 +44,13 @@ class Scanner(object):
self.bus.set_flushing(True)
def scan(self, uri):
"""
Scan the given uri collecting relevant metadata.
:param uri: URI of the resource to scan.
:type event: string
:return: Dictionary of tags, duration, mtime and uri information.
"""
try:
self._setup(uri)
data = self._collect()

4
mopidy/backends/local/__init__.py
View File

@ -36,3 +36,7 @@ class Extension(ext.Extension):
def get_library_updaters(self):
from .library import LocalLibraryUpdateProvider
return [LocalLibraryUpdateProvider]
def get_command(self):
from .commands import LocalCommand
return LocalCommand()

115
mopidy/backends/local/commands.py Normal file
View File

@ -0,0 +1,115 @@
from __future__ import unicode_literals
import logging
import os
import time
from mopidy import commands, exceptions
from mopidy.audio import scan
from mopidy.utils import path
from . import translator
logger = logging.getLogger('mopidy.backends.local.commands')
class LocalCommand(commands.Command):
def __init__(self):
super(LocalCommand, self).__init__()
self.add_child('scan', ScanCommand())
class ScanCommand(commands.Command):
help = "Scan local media files and populate the local library."
def run(self, args, config, extensions):
media_dir = config['local']['media_dir']
scan_timeout = config['local']['scan_timeout']
excluded_file_extensions = set(
ext.lower() for ext in config['local']['excluded_file_extensions'])
updaters = {}
for e in extensions:
for updater_class in e.get_library_updaters():
if updater_class and 'local' in updater_class.uri_schemes:
updaters[e.ext_name] = updater_class
if not updaters:
logger.error('No usable library updaters found.')
return 1
elif len(updaters) > 1:
logger.error('More than one library updater found. '
'Provided by: %s', ', '.join(updaters.keys()))
return 1
local_updater = updaters.values()[0](config)
# TODO: cleanup to consistently use local urls, not a random mix of
# local and file uris depending on how the data was loaded.
uris_library = set()
uris_update = set()
uris_remove = set()
tracks = local_updater.load()
logger.info('Checking %d tracks from library.', len(tracks))
for track in tracks:
try:
uri = translator.local_to_file_uri(track.uri, media_dir)
stat = os.stat(path.uri_to_path(uri))
if int(stat.st_mtime) > track.last_modified:
uris_update.add(uri)
uris_library.add(uri)
except OSError:
logger.debug('Missing file %s', track.uri)
uris_remove.add(track.uri)
logger.info('Removing %d missing tracks.', len(uris_remove))
for uri in uris_remove:
local_updater.remove(uri)
logger.info('Checking %s for unknown tracks.', media_dir)
for uri in path.find_uris(media_dir):
file_extension = os.path.splitext(path.uri_to_path(uri))[1]
if file_extension.lower() in excluded_file_extensions:
logger.debug('Skipped %s: File extension excluded.', uri)
continue
if uri not in uris_library:
uris_update.add(uri)
logger.info('Found %d unknown tracks.', len(uris_update))
logger.info('Scanning...')
scanner = scan.Scanner(scan_timeout)
progress = Progress(len(uris_update))
for uri in sorted(uris_update):
try:
data = scanner.scan(uri)
track = scan.audio_data_to_track(data)
local_updater.add(track)
logger.debug('Added %s', track.uri)
except exceptions.ScannerError as error:
logger.warning('Failed %s: %s', uri, error)
progress.increment()
logger.info('Commiting changes.')
local_updater.commit()
return 0
# TODO: move to utils?
class Progress(object):
def __init__(self, total):
self.count = 0
self.total = total
self.start = time.time()
def increment(self):
self.count += 1
if self.count % 1000 == 0 or self.count == self.total:
duration = time.time() - self.start
remainder = duration / self.count * (self.total - self.count)
logger.info('Scanned %d of %d files in %ds, ~%ds left.',
self.count, self.total, duration, remainder)

383
mopidy/commands.py
View File

@ -1,10 +1,18 @@
from __future__ import unicode_literals
import argparse
import collections
import gobject
import logging
import os
import sys
from mopidy import config as config_lib, ext
from mopidy.utils import deps, versioning
from mopidy import config as config_lib
from mopidy.audio import Audio
from mopidy.core import Core
from mopidy.utils import deps, process, versioning
logger = logging.getLogger('mopidy.commands')
def config_files_type(value):
@ -21,63 +29,322 @@ def config_override_type(value):
'%s must have the format section/key=value' % value)
parser = argparse.ArgumentParser()
parser.add_argument(
'--version', action='version',
version='Mopidy %s' % versioning.get_version())
parser.add_argument(
'-q', '--quiet',
action='store_const', const=-1, dest='verbosity_level',
help='less output (warning level)')
parser.add_argument(
'-v', '--verbose',
action='count', dest='verbosity_level',
help='more output (debug level)')
parser.add_argument(
'--save-debug-log',
action='store_true', dest='save_debug_log',
help='save debug log to "./mopidy.log"')
parser.add_argument(
'--show-config',
action='store_true', dest='show_config',
help='show current config')
parser.add_argument(
'--show-deps',
action='store_true', dest='show_deps',
help='show dependencies and their versions')
parser.add_argument(
'--config',
action='store', dest='config_files', type=config_files_type,
default=b'$XDG_CONFIG_DIR/mopidy/mopidy.conf',
help='config files to use, colon seperated, later files override')
parser.add_argument(
'-o', '--option',
action='append', dest='config_overrides', type=config_override_type,
help='`section/key=value` values to override config options')
class _ParserError(Exception):
pass
def show_config(args):
"""Prints the effective config and exits."""
extensions = ext.load_extensions()
config, errors = config_lib.load(
args.config_files, extensions, args.config_overrides)
# Clear out any config for disabled extensions.
for extension in extensions:
if not ext.validate_extension(extension):
config[extension.ext_name] = {b'enabled': False}
errors[extension.ext_name] = {
b'enabled': b'extension disabled itself.'}
elif not config[extension.ext_name]['enabled']:
config[extension.ext_name] = {b'enabled': False}
errors[extension.ext_name] = {
b'enabled': b'extension disabled by config.'}
print config_lib.format(config, extensions, errors)
sys.exit(0)
class _HelpError(Exception):
pass
def show_deps():
"""Prints a list of all dependencies and exits."""
print deps.format_dependency_list()
sys.exit(0)
class _ArgumentParser(argparse.ArgumentParser):
def error(self, message):
raise _ParserError(message)
class _HelpAction(argparse.Action):
def __init__(self, option_strings, dest=None, help=None):
super(_HelpAction, self).__init__(
option_strings=option_strings,
dest=dest or argparse.SUPPRESS,
default=argparse.SUPPRESS,
nargs=0,
help=help)
def __call__(self, parser, namespace, values, option_string=None):
raise _HelpError()
class Command(object):
"""Command parser and runner for building trees of commands.
This class provides a wraper around :class:`argparse.ArgumentParser`
for handling this type of command line application in a better way than
argprases own sub-parser handling.
"""
help = None
#: Help text to display in help output.
def __init__(self):
self._children = collections.OrderedDict()
self._arguments = []
self._overrides = {}
def _build(self):
actions = []
parser = _ArgumentParser(add_help=False)
parser.register('action', 'help', _HelpAction)
for args, kwargs in self._arguments:
actions.append(parser.add_argument(*args, **kwargs))
parser.add_argument('_args', nargs=argparse.REMAINDER,
help=argparse.SUPPRESS)
return parser, actions
def add_child(self, name, command):
"""Add a child parser to consider using.
:param name: name to use for the sub-command that is being added.
:type name: string
"""
self._children[name] = command
def add_argument(self, *args, **kwargs):
"""Add am argument to the parser.
This method takes all the same arguments as the
:class:`argparse.ArgumentParser` version of this method.
"""
self._arguments.append((args, kwargs))
def set(self, **kwargs):
"""Override a value in the finaly result of parsing."""
self._overrides.update(kwargs)
def exit(self, status_code=0, message=None, usage=None):
"""Optionally print a message and exit."""
print '\n\n'.join(m for m in (usage, message) if m)
sys.exit(status_code)
def format_usage(self, prog=None):
"""Format usage for current parser."""
actions = self._build()[1]
prog = prog or os.path.basename(sys.argv[0])
return self._usage(actions, prog) + '\n'
def _usage(self, actions, prog):
formatter = argparse.HelpFormatter(prog)
formatter.add_usage(None, actions, [])
return formatter.format_help().strip()
def format_help(self, prog=None):
"""Format help for current parser and children."""
actions = self._build()[1]
prog = prog or os.path.basename(sys.argv[0])
formatter = argparse.HelpFormatter(prog)
formatter.add_usage(None, actions, [])
if self.help:
formatter.add_text(self.help)
if actions:
formatter.add_text('OPTIONS:')
formatter.start_section(None)
formatter.add_arguments(actions)
formatter.end_section()
subhelp = []
for name, child in self._children.items():
child._subhelp(name, subhelp)
if subhelp:
formatter.add_text('COMMANDS:')
subhelp.insert(0, '')
return formatter.format_help() + '\n'.join(subhelp)
def _subhelp(self, name, result):
actions = self._build()[1]
if self.help or actions:
formatter = argparse.HelpFormatter(name)
formatter.add_usage(None, actions, [], '')
formatter.start_section(None)
formatter.add_text(self.help)
formatter.start_section(None)
formatter.add_arguments(actions)
formatter.end_section()
formatter.end_section()
result.append(formatter.format_help())
for childname, child in self._children.items():
child._subhelp(' '.join((name, childname)), result)
def parse(self, args, prog=None):
"""Parse command line arguments.
Will recursively parse commands until a final parser is found or an
error occurs. In the case of errors we will print a message and exit.
Otherwise, any overrides are applied and the current parser stored
in the command attribute of the return value.
:param args: list of arguments to parse
:type args: list of strings
:param prog: name to use for program
:type prog: string
:rtype: :class:`argparse.Namespace`
"""
prog = prog or os.path.basename(sys.argv[0])
try:
return self._parse(
args, argparse.Namespace(), self._overrides.copy(), prog)
except _HelpError:
self.exit(0, self.format_help(prog))
def _parse(self, args, namespace, overrides, prog):
overrides.update(self._overrides)
parser, actions = self._build()
try:
result = parser.parse_args(args, namespace)
except _ParserError as e:
self.exit(1, e.message, self._usage(actions, prog))
if not result._args:
for attr, value in overrides.items():
setattr(result, attr, value)
delattr(result, '_args')
result.command = self
return result
child = result._args.pop(0)
if child not in self._children:
usage = self._usage(actions, prog)
self.exit(1, 'unrecognized command: %s' % child, usage)
return self._children[child]._parse(
result._args, result, overrides, ' '.join([prog, child]))
def run(self, *args, **kwargs):
"""Run the command.
Must be implemented by sub-classes that are not simply and intermediate
in the command namespace.
"""
raise NotImplementedError
class RootCommand(Command):
def __init__(self):
super(RootCommand, self).__init__()
self.set(base_verbosity_level=0)
self.add_argument(
'-h', '--help',
action='help', help='Show this message and exit')
self.add_argument(
'--version', action='version',
version='Mopidy %s' % versioning.get_version())
self.add_argument(
'-q', '--quiet',
action='store_const', const=-1, dest='verbosity_level',
help='less output (warning level)')
self.add_argument(
'-v', '--verbose',
action='count', dest='verbosity_level', default=0,
help='more output (debug level)')
self.add_argument(
'--save-debug-log',
action='store_true', dest='save_debug_log',
help='save debug log to "./mopidy.log"')
self.add_argument(
'--config',
action='store', dest='config_files', type=config_files_type,
default=b'$XDG_CONFIG_DIR/mopidy/mopidy.conf', metavar='FILES',
help='config files to use, colon seperated, later files override')
self.add_argument(
'-o', '--option',
action='append', dest='config_overrides',
type=config_override_type, metavar='OPTIONS',
help='`section/key=value` values to override config options')
def run(self, args, config, extensions):
loop = gobject.MainLoop()
try:
audio = self.start_audio(config)
backends = self.start_backends(config, extensions, audio)
core = self.start_core(audio, backends)
self.start_frontends(config, extensions, core)
loop.run()
except KeyboardInterrupt:
logger.info('Interrupted. Exiting...')
return
finally:
loop.quit()
self.stop_frontends(extensions)
self.stop_core()
self.stop_backends(extensions)
self.stop_audio()
process.stop_remaining_actors()
def start_audio(self, config):
logger.info('Starting Mopidy audio')
return Audio.start(config=config).proxy()
def start_backends(self, config, extensions, audio):
backend_classes = []
for extension in extensions:
backend_classes.extend(extension.get_backend_classes())
logger.info(
'Starting Mopidy backends: %s',
', '.join(b.__name__ for b in backend_classes) or 'none')
backends = []
for backend_class in backend_classes:
backend = backend_class.start(config=config, audio=audio).proxy()
backends.append(backend)
return backends
def start_core(self, audio, backends):
logger.info('Starting Mopidy core')
return Core.start(audio=audio, backends=backends).proxy()
def start_frontends(self, config, extensions, core):
frontend_classes = []
for extension in extensions:
frontend_classes.extend(extension.get_frontend_classes())
logger.info(
'Starting Mopidy frontends: %s',
', '.join(f.__name__ for f in frontend_classes) or 'none')
for frontend_class in frontend_classes:
frontend_class.start(config=config, core=core)
def stop_frontends(self, extensions):
logger.info('Stopping Mopidy frontends')
for extension in extensions:
for frontend_class in extension.get_frontend_classes():
process.stop_actors_by_class(frontend_class)
def stop_core(self):
logger.info('Stopping Mopidy core')
process.stop_actors_by_class(Core)
def stop_backends(self, extensions):
logger.info('Stopping Mopidy backends')
for extension in extensions:
for backend_class in extension.get_backend_classes():
process.stop_actors_by_class(backend_class)
def stop_audio(self):
logger.info('Stopping Mopidy audio')
process.stop_actors_by_class(Audio)
class ConfigCommand(Command):
help = "Show currently active configuration."
def __init__(self):
super(ConfigCommand, self).__init__()
self.set(base_verbosity_level=-1)
def run(self, config, errors, extensions):
print config_lib.format(config, extensions, errors)
return 0
class DepsCommand(Command):
help = "Show dependencies and debug information."
def __init__(self):
super(DepsCommand, self).__init__()
self.set(base_verbosity_level=-1)
def run(self):
print deps.format_dependency_list()
return 0

8
mopidy/ext.py
View File

@ -87,6 +87,14 @@ class Extension(object):
"""
return []
def get_command(self):
"""Command to expose to command line users running mopidy.
:returns:
Instance of a :class:`~mopidy.commands.Command` class.
"""
pass
def register_gstreamer_elements(self):
"""Hook for registering custom GStreamer elements

157
mopidy/scanner.py
View File

@ -1,157 +0,0 @@
from __future__ import unicode_literals
import argparse
import logging
import os
import sys
import time
import gobject
gobject.threads_init()
# Extract any command line arguments. This needs to be done before GStreamer is
# imported, so that GStreamer doesn't hijack e.g. ``--help``.
mopidy_args = sys.argv[1:]
sys.argv[1:] = []
from mopidy import config as config_lib, exceptions, ext
from mopidy.audio import scan
from mopidy.backends.local import translator
from mopidy.utils import log, path, versioning
def main():
args = parse_args()
# TODO: support config files and overrides (shared from main?)
config_files = [b'/etc/mopidy/mopidy.conf',
b'$XDG_CONFIG_DIR/mopidy/mopidy.conf']
config_overrides = []
# TODO: decide if we want to avoid this boilerplate some how.
# Initial config without extensions to bootstrap logging.
logging_config, _ = config_lib.load(config_files, [], config_overrides)
log.setup_root_logger()
log.setup_console_logging(logging_config, args.verbosity_level)
extensions = ext.load_extensions()
config, errors = config_lib.load(
config_files, extensions, config_overrides)
log.setup_log_levels(config)
if not config['local']['media_dir']:
logging.warning('Config value local/media_dir is not set.')
return
if not config['local']['scan_timeout']:
logging.warning('Config value local/scan_timeout is not set.')
return
# TODO: missing config error checking and other default setup code.
updaters = {}
for e in extensions:
for updater_class in e.get_library_updaters():
if updater_class and 'local' in updater_class.uri_schemes:
updaters[e.ext_name] = updater_class
if not updaters:
logging.error('No usable library updaters found.')
return
elif len(updaters) > 1:
logging.error('More than one library updater found. '
'Provided by: %s', ', '.join(updaters.keys()))
return
local_updater = updaters.values()[0](config) # TODO: switch to actor?
media_dir = config['local']['media_dir']
excluded_extensions = set(
file_ext.lower()
for file_ext in config['local']['excluded_file_extensions'])
# TODO: cleanup to consistently use local urls, not a random mix of local
# and file uris depending on how the data was loaded.
uris_library = set()
uris_update = set()
uris_remove = set()
logging.info('Checking tracks from library.')
for track in local_updater.load():
try:
uri = translator.local_to_file_uri(track.uri, media_dir)
stat = os.stat(path.uri_to_path(uri))
if int(stat.st_mtime) > track.last_modified:
uris_update.add(uri)
uris_library.add(uri)
except OSError:
logging.debug('Missing file %s', track.uri)
uris_remove.add(track.uri)
logging.info('Removing %d missing tracks.', len(uris_remove))
for uri in uris_remove:
local_updater.remove(uri)
logging.info('Checking %s for unknown tracks.', media_dir)
for uri in path.find_uris(config['local']['media_dir']):
file_ext = os.path.splitext(path.uri_to_path(uri))[1]
if file_ext.lower() in excluded_extensions:
logging.debug('Skipped %s: File extension excluded.', uri)
continue
if uri not in uris_library:
uris_update.add(uri)
logging.info('Found %d unknown tracks.', len(uris_update))
logging.info('Scanning...')
scanner = scan.Scanner(config['local']['scan_timeout'])
progress = Progress(len(uris_update))
for uri in sorted(uris_update):
try:
data = scanner.scan(uri)
track = scan.audio_data_to_track(data)
local_updater.add(track)
logging.debug('Added %s', track.uri)
except exceptions.ScannerError as error:
logging.warning('Failed %s: %s', uri, error)
progress.increment()
logging.info('Commiting changes.')
local_updater.commit()
class Progress(object):
def __init__(self, total):
self.count = 0
self.total = total
self.start = time.time()
def increment(self):
self.count += 1
if self.count % 1000 == 0 or self.count == self.total:
duration = time.time() - self.start
remainder = duration / self.count * (self.total - self.count)
logging.info('Scanned %d of %d files in %ds, ~%ds left.',
self.count, self.total, duration, remainder)
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument(
'--version', action='version',
version='Mopidy %s' % versioning.get_version())
parser.add_argument(
'-q', '--quiet',
action='store_const', const=0, dest='verbosity_level',
help='less output (warning level)')
parser.add_argument(
'-v', '--verbose',
action='count', default=1, dest='verbosity_level',
help='more output (debug level)')
return parser.parse_args(args=mopidy_args)
if __name__ == '__main__':
main()

39
mopidy/utils/log.py
View File

@ -4,12 +4,37 @@ import logging
import logging.config
import logging.handlers
from . import versioning
class DelayedHandler(logging.Handler):
def __init__(self):
logging.Handler.__init__(self)
self._released = False
self._buffer = []
def handle(self, record):
if not self._released:
self._buffer.append(record)
def release(self):
self._released = True
root = logging.getLogger('')
while self._buffer:
root.handle(self._buffer.pop(0))
_delayed_handler = DelayedHandler()
def bootstrap_delayed_logging():
root = logging.getLogger('')
root.setLevel(logging.DEBUG)
root.addHandler(_delayed_handler)
def setup_logging(config, verbosity_level, save_debug_log):
setup_root_logger()
setup_console_logging(config, verbosity_level)
setup_log_levels(config)
if save_debug_log:
setup_debug_logging_to_file(config)
@ -18,8 +43,7 @@ def setup_logging(config, verbosity_level, save_debug_log):
if config['logging']['config_file']:
logging.config.fileConfig(config['logging']['config_file'])
logger = logging.getLogger('mopidy.utils.log')
logger.info('Starting Mopidy %s', versioning.get_version())
_delayed_handler.release()
def setup_log_levels(config):
@ -27,13 +51,8 @@ def setup_log_levels(config):
logging.getLogger(name).setLevel(level)
def setup_root_logger():
root = logging.getLogger('')
root.setLevel(logging.DEBUG)
def setup_console_logging(config, verbosity_level):
if verbosity_level == -1:
if verbosity_level < 0:
log_level = logging.WARNING
log_format = config['logging']['console_format']
elif verbosity_level >= 1:

1
setup.py
View File

@ -38,7 +38,6 @@ setup(
entry_points={
'console_scripts': [
'mopidy = mopidy.__main__:main',
'mopidy-scan = mopidy.scanner:main',
'mopidy-convert-config = mopidy.config.convert:main',
],
'mopidy.ext': [

448
tests/commands_test.py
View File

@ -1,6 +1,7 @@
from __future__ import unicode_literals
import argparse
import mock
import unittest
from mopidy import commands
@ -42,3 +43,450 @@ class ConfigOverrideTypeTest(unittest.TestCase):
self.assertRaises(
argparse.ArgumentTypeError,
commands.config_override_type, b'section')
class CommandParsingTest(unittest.TestCase):
def setUp(self):
self.exit_patcher = mock.patch.object(commands.Command, 'exit')
self.exit_mock = self.exit_patcher.start()
self.exit_mock.side_effect = SystemExit
def tearDown(self):
self.exit_patcher.stop()
def test_command_parsing_returns_namespace(self):
cmd = commands.Command()
self.assertIsInstance(cmd.parse([]), argparse.Namespace)
def test_command_parsing_does_not_contain_args(self):
cmd = commands.Command()
result = cmd.parse([])
self.assertFalse(hasattr(result, '_args'))
def test_unknown_options_bails(self):
cmd = commands.Command()
with self.assertRaises(SystemExit):
cmd.parse(['--foobar'])
def test_invalid_sub_command_bails(self):
cmd = commands.Command()
with self.assertRaises(SystemExit):
cmd.parse(['foo'])
def test_command_arguments(self):
cmd = commands.Command()
cmd.add_argument('--bar')
result = cmd.parse(['--bar', 'baz'])
self.assertEqual(result.bar, 'baz')
def test_command_arguments_and_sub_command(self):
child = commands.Command()
child.add_argument('--baz')
cmd = commands.Command()
cmd.add_argument('--bar')
cmd.add_child('foo', child)
result = cmd.parse(['--bar', 'baz', 'foo'])
self.assertEqual(result.bar, 'baz')
self.assertEqual(result.baz, None)
def test_subcommand_may_have_positional(self):
child = commands.Command()
child.add_argument('bar')
cmd = commands.Command()
cmd.add_child('foo', child)
result = cmd.parse(['foo', 'baz'])
self.assertEqual(result.bar, 'baz')
def test_subcommand_may_have_remainder(self):
child = commands.Command()
child.add_argument('bar', nargs=argparse.REMAINDER)
cmd = commands.Command()
cmd.add_child('foo', child)
result = cmd.parse(['foo', 'baz', 'bep', 'bop'])
self.assertEqual(result.bar, ['baz', 'bep', 'bop'])
def test_result_stores_choosen_command(self):
child = commands.Command()
cmd = commands.Command()
cmd.add_child('foo', child)
result = cmd.parse(['foo'])
self.assertEqual(result.command, child)
result = cmd.parse([])
self.assertEqual(result.command, cmd)
child2 = commands.Command()
cmd.add_child('bar', child2)
subchild = commands.Command()
child.add_child('baz', subchild)
result = cmd.parse(['bar'])
self.assertEqual(result.command, child2)
result = cmd.parse(['foo', 'baz'])
self.assertEqual(result.command, subchild)
def test_invalid_type(self):
cmd = commands.Command()
cmd.add_argument('--bar', type=int)
with self.assertRaises(SystemExit):
cmd.parse(['--bar', b'zero'], prog='foo')
self.exit_mock.assert_called_once_with(
1, "argument --bar: invalid int value: 'zero'",
'usage: foo [--bar BAR]')
@mock.patch('sys.argv')
def test_command_error_usage_prog(self, argv_mock):
argv_mock.__getitem__.return_value = '/usr/bin/foo'
cmd = commands.Command()
cmd.add_argument('--bar', required=True)
with self.assertRaises(SystemExit):
cmd.parse([])
self.exit_mock.assert_called_once_with(
mock.ANY, mock.ANY, 'usage: foo --bar BAR')
self.exit_mock.reset_mock()
with self.assertRaises(SystemExit):
cmd.parse([], prog='baz')
self.exit_mock.assert_called_once_with(
mock.ANY, mock.ANY, 'usage: baz --bar BAR')
def test_missing_required(self):
cmd = commands.Command()
cmd.add_argument('--bar', required=True)
with self.assertRaises(SystemExit):
cmd.parse([], prog='foo')
self.exit_mock.assert_called_once_with(
1, 'argument --bar is required', 'usage: foo --bar BAR')
def test_missing_positionals(self):
cmd = commands.Command()
cmd.add_argument('bar')
with self.assertRaises(SystemExit):
cmd.parse([], prog='foo')
self.exit_mock.assert_called_once_with(
1, 'too few arguments', 'usage: foo bar')
def test_missing_positionals_subcommand(self):
child = commands.Command()
child.add_argument('baz')
cmd = commands.Command()
cmd.add_child('bar', child)
with self.assertRaises(SystemExit):
cmd.parse(['bar'], prog='foo')
self.exit_mock.assert_called_once_with(
1, 'too few arguments', 'usage: foo bar baz')
def test_unknown_command(self):
cmd = commands.Command()
with self.assertRaises(SystemExit):
cmd.parse(['--help'], prog='foo')
self.exit_mock.assert_called_once_with(
1, 'unrecognized arguments: --help', 'usage: foo')
def test_invalid_subcommand(self):
cmd = commands.Command()
cmd.add_child('baz', commands.Command())
with self.assertRaises(SystemExit):
cmd.parse(['bar'], prog='foo')
self.exit_mock.assert_called_once_with(
1, 'unrecognized command: bar', 'usage: foo')
def test_set(self):
cmd = commands.Command()
cmd.set(foo='bar')
result = cmd.parse([])
self.assertEqual(result.foo, 'bar')
def test_set_propegate(self):
child = commands.Command()
cmd = commands.Command()
cmd.set(foo='bar')
cmd.add_child('command', child)
result = cmd.parse(['command'])
self.assertEqual(result.foo, 'bar')
def test_innermost_set_wins(self):
child = commands.Command()
child.set(foo='bar', baz=1)
cmd = commands.Command()
cmd.set(foo='baz', baz=None)
cmd.add_child('command', child)
result = cmd.parse(['command'])
self.assertEqual(result.foo, 'bar')
self.assertEqual(result.baz, 1)
def test_help_action_works(self):
cmd = commands.Command()
cmd.add_argument('-h', action='help')
cmd.format_help = mock.Mock()
with self.assertRaises(SystemExit):
cmd.parse(['-h'])
cmd.format_help.assert_called_once_with(mock.ANY)
self.exit_mock.assert_called_once_with(0, cmd.format_help.return_value)
class UsageTest(unittest.TestCase):
@mock.patch('sys.argv')
def test_prog_name_default_and_override(self, argv_mock):
argv_mock.__getitem__.return_value = '/usr/bin/foo'
cmd = commands.Command()
self.assertEqual('usage: foo', cmd.format_usage().strip())
self.assertEqual('usage: baz', cmd.format_usage('baz').strip())
def test_basic_usage(self):
cmd = commands.Command()
self.assertEqual('usage: foo', cmd.format_usage('foo').strip())
cmd.add_argument('-h', '--help', action='store_true')
self.assertEqual('usage: foo [-h]', cmd.format_usage('foo').strip())
cmd.add_argument('bar')
self.assertEqual('usage: foo [-h] bar',
cmd.format_usage('foo').strip())
def test_nested_usage(self):
child = commands.Command()
cmd = commands.Command()
cmd.add_child('bar', child)
self.assertEqual('usage: foo', cmd.format_usage('foo').strip())
self.assertEqual('usage: foo bar', cmd.format_usage('foo bar').strip())
cmd.add_argument('-h', '--help', action='store_true')
self.assertEqual('usage: foo bar',
child.format_usage('foo bar').strip())
child.add_argument('-h', '--help', action='store_true')
self.assertEqual('usage: foo bar [-h]',
child.format_usage('foo bar').strip())
class HelpTest(unittest.TestCase):
@mock.patch('sys.argv')
def test_prog_name_default_and_override(self, argv_mock):
argv_mock.__getitem__.return_value = '/usr/bin/foo'
cmd = commands.Command()
self.assertEqual('usage: foo', cmd.format_help().strip())
self.assertEqual('usage: bar', cmd.format_help('bar').strip())
def test_command_without_documenation_or_options(self):
cmd = commands.Command()
self.assertEqual('usage: bar', cmd.format_help('bar').strip())
def test_command_with_option(self):
cmd = commands.Command()
cmd.add_argument('-h', '--help', action='store_true',
help='show this message')
expected = ('usage: foo [-h]\n\n'
'OPTIONS:\n\n'
' -h, --help show this message')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_command_with_option_and_positional(self):
cmd = commands.Command()
cmd.add_argument('-h', '--help', action='store_true',
help='show this message')
cmd.add_argument('bar', help='some help text')
expected = ('usage: foo [-h] bar\n\n'
'OPTIONS:\n\n'
' -h, --help show this message\n'
' bar some help text')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_command_with_documentation(self):
cmd = commands.Command()
cmd.help = 'some text about everything this command does.'
expected = ('usage: foo\n\n'
'some text about everything this command does.')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_command_with_documentation_and_option(self):
cmd = commands.Command()
cmd.help = 'some text about everything this command does.'
cmd.add_argument('-h', '--help', action='store_true',
help='show this message')
expected = ('usage: foo [-h]\n\n'
'some text about everything this command does.\n\n'
'OPTIONS:\n\n'
' -h, --help show this message')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_subcommand_without_documentation_or_options(self):
child = commands.Command()
cmd = commands.Command()
cmd.add_child('bar', child)
self.assertEqual('usage: foo', cmd.format_help('foo').strip())
def test_subcommand_with_documentation_shown(self):
child = commands.Command()
child.help = 'some text about everything this command does.'
cmd = commands.Command()
cmd.add_child('bar', child)
expected = ('usage: foo\n\n'
'COMMANDS:\n\n'
'bar\n\n'
' some text about everything this command does.')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_subcommand_with_options_shown(self):
child = commands.Command()
child.add_argument('-h', '--help', action='store_true',
help='show this message')
cmd = commands.Command()
cmd.add_child('bar', child)
expected = ('usage: foo\n\n'
'COMMANDS:\n\n'
'bar [-h]\n\n'
' -h, --help show this message')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_subcommand_with_positional_shown(self):
child = commands.Command()
child.add_argument('baz', help='the great and wonderful')
cmd = commands.Command()
cmd.add_child('bar', child)
expected = ('usage: foo\n\n'
'COMMANDS:\n\n'
'bar baz\n\n'
' baz the great and wonderful')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_subcommand_with_options_and_documentation(self):
child = commands.Command()
child.help = ' some text about everything this command does.'
child.add_argument('-h', '--help', action='store_true',
help='show this message')
cmd = commands.Command()
cmd.add_child('bar', child)
expected = ('usage: foo\n\n'
'COMMANDS:\n\n'
'bar [-h]\n\n'
' some text about everything this command does.\n\n'
' -h, --help show this message')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_nested_subcommands_with_options(self):
subchild = commands.Command()
subchild.add_argument('--test', help='the great and wonderful')
child = commands.Command()
child.add_child('baz', subchild)
child.add_argument('-h', '--help', action='store_true',
help='show this message')
cmd = commands.Command()
cmd.add_child('bar', child)
expected = ('usage: foo\n\n'
'COMMANDS:\n\n'
'bar [-h]\n\n'
' -h, --help show this message\n\n'
'bar baz [--test TEST]\n\n'
' --test TEST the great and wonderful')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_nested_subcommands_skipped_intermediate(self):
subchild = commands.Command()
subchild.add_argument('--test', help='the great and wonderful')
child = commands.Command()
child.add_child('baz', subchild)
cmd = commands.Command()
cmd.add_child('bar', child)
expected = ('usage: foo\n\n'
'COMMANDS:\n\n'
'bar baz [--test TEST]\n\n'
' --test TEST the great and wonderful')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_command_with_option_and_subcommand_with_option(self):
child = commands.Command()
child.add_argument('--test', help='the great and wonderful')
cmd = commands.Command()
cmd.add_argument('-h', '--help', action='store_true',
help='show this message')
cmd.add_child('bar', child)
expected = ('usage: foo [-h]\n\n'
'OPTIONS:\n\n'
' -h, --help show this message\n\n'
'COMMANDS:\n\n'
'bar [--test TEST]\n\n'
' --test TEST the great and wonderful')
self.assertEqual(expected, cmd.format_help('foo').strip())
def test_command_with_options_doc_and_subcommand_with_option_and_doc(self):
child = commands.Command()
child.help = 'some text about this sub-command.'
child.add_argument('--test', help='the great and wonderful')
cmd = commands.Command()
cmd.help = 'some text about everything this command does.'
cmd.add_argument('-h', '--help', action='store_true',
help='show this message')
cmd.add_child('bar', child)
expected = ('usage: foo [-h]\n\n'
'some text about everything this command does.\n\n'
'OPTIONS:\n\n'
' -h, --help show this message\n\n'
'COMMANDS:\n\n'
'bar [--test TEST]\n\n'
' some text about this sub-command.\n\n'
' --test TEST the great and wonderful')
self.assertEqual(expected, cmd.format_help('foo').strip())
class RunTest(unittest.TestCase):
def test_default_implmentation_raises_error(self):
with self.assertRaises(NotImplementedError):
commands.Command().run()

1
tests/help_test.py
View File

@ -22,6 +22,5 @@ class HelpTest(unittest.TestCase):
self.assertIn('--quiet', output)
self.assertIn('--verbose', output)
self.assertIn('--save-debug-log', output)
self.assertIn('--show-config', output)
self.assertIn('--config', output)
self.assertIn('--option', output)