pw_cli: Generalize the plugin system
- Move the global plugin variables into a class. This makes it possible
to use multiple plugin registries, potentially for different purposes.
- Add tests for the pw.plugins.Registry class.
- Expose the Plugin class. This allows using plugins for ways
other than as command line utilities.
- Always include __init__ functions in Sphinx autodoc.
Change-Id: Id8aa40234f81c1da46ec51e5590b16c561934be3
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/40761
Commit-Queue: Wyatt Hepler <hepler@google.com>
Pigweed-Auto-Submit: Wyatt Hepler <hepler@google.com>
Reviewed-by: Keir Mierle <keir@google.com>
Reviewed-by: Rob Mohr <mohrr@google.com>
diff --git a/docs/conf.py b/docs/conf.py
index 1885a99..7eaf374 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -114,3 +114,14 @@
# Markdown files imported using m2r aren't marked as "referenced," so exclude
# them from the error reference checking.
exclude_patterns = ['README.md']
+
+
+def do_not_skip_init(app, what, name, obj, would_skip, options):
+ if name == "__init__":
+ return False # never skip __init__ functions
+
+ return would_skip
+
+
+def setup(app):
+ app.connect("autodoc-skip-member", do_not_skip_init)
diff --git a/pw_cli/docs.rst b/pw_cli/docs.rst
index 40489a2..35378e0 100644
--- a/pw_cli/docs.rst
+++ b/pw_cli/docs.rst
@@ -271,7 +271,40 @@
The ``pw_cli`` Pigweed module includes the ``pw_cli`` Python package, which
provides utilities for creating command line tools with Pigweed.
-pw_cli log
+pw_cli.log
----------
.. automodule:: pw_cli.log
:members:
+
+pw_cli.plugins
+--------------
+:py:mod:`pw_cli.plugins` provides general purpose plugin functionality. The
+module can be used to create plugins for command line tools, interactive
+consoles, or anything else. Pigweed's ``pw`` command uses this module for its
+plugins.
+
+To use plugins, create a :py:class:`pw_cli.plugins.Registry`. The registry may
+have an optional validator function that checks plugins before they are
+registered (see :py:meth:`pw_cli.plugins.Registry.__init__`).
+
+Plugins may be registered in a few different ways.
+
+ * Register with a direct function call. See
+ :py:meth:`pw_cli.plugins.Registry.register` and
+ :py:meth:`pw_cli.plugins.Registry.register_by_name`.
+ * Register from plugins files. See
+ :py:meth:`pw_cli.plugins.Registry.register_file` and
+ :py:meth:`pw_cli.plugins.Registry.register_directory`. Plugins files use a
+ simple format:
+
+ .. code-block::
+
+ # Comments start with "#". Blank lines are ignored.
+ name_of_the_plugin module.name module_member
+
+ another_plugin some_module some_function
+
+Module reference
+^^^^^^^^^^^^^^^^
+.. automodule:: pw_cli.plugins
+ :members:
diff --git a/pw_cli/py/BUILD.gn b/pw_cli/py/BUILD.gn
index 2f04bde..b7ed63c 100644
--- a/pw_cli/py/BUILD.gn
+++ b/pw_cli/py/BUILD.gn
@@ -29,8 +29,12 @@
"pw_cli/log.py",
"pw_cli/plugins.py",
"pw_cli/process.py",
+ "pw_cli/pw_command_plugins.py",
"pw_cli/requires.py",
]
- tests = [ "envparse_test.py" ]
+ tests = [
+ "envparse_test.py",
+ "plugins_test.py",
+ ]
pylintrc = "$dir_pigweed/.pylintrc"
}
diff --git a/pw_cli/py/plugins_test.py b/pw_cli/py/plugins_test.py
new file mode 100644
index 0000000..908bc4b
--- /dev/null
+++ b/pw_cli/py/plugins_test.py
@@ -0,0 +1,189 @@
+# Copyright 2021 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""Tests for pw_cli.plugins."""
+
+from pathlib import Path
+import sys
+import tempfile
+import types
+from typing import Dict, Iterator
+import unittest
+
+from pw_cli import plugins
+
+
+def _no_docstring() -> int:
+ return 123
+
+
+def _with_docstring() -> int:
+ """This docstring is brought to you courtesy of Pigweed."""
+ return 456
+
+
+def _create_files(directory: str, files: Dict[str, str]) -> Iterator[Path]:
+ for relative_path, contents in files.items():
+ path = Path(directory) / relative_path
+ path.parent.mkdir(exist_ok=True, parents=True)
+ path.write_text(contents)
+ yield path
+
+
+class TestPlugin(unittest.TestCase):
+ """Tests for plugins.Plugins."""
+ def test_target_name_attribute(self) -> None:
+ self.assertEqual(
+ plugins.Plugin('abc', _no_docstring).target_name,
+ f'{__name__}._no_docstring')
+
+ def test_target_name_no_name_attribute(self) -> None:
+ has_no_name = 'no __name__'
+ self.assertFalse(hasattr(has_no_name, '__name__'))
+
+ self.assertEqual(
+ plugins.Plugin('abc', has_no_name).target_name,
+ '<unknown>.no __name__')
+
+
+_TEST_PLUGINS = {
+ 'TEST_PLUGINS': (f'test_plugin {__name__} _with_docstring\n'
+ f'other_plugin {__name__} _no_docstring\n'),
+ 'nested/in/dirs/TEST_PLUGINS':
+ f'test_plugin {__name__} _no_docstring\n',
+}
+
+
+class TestPluginRegistry(unittest.TestCase):
+ """Tests for plugins.Registry."""
+ def setUp(self) -> None:
+ self._registry = plugins.Registry(
+ validator=plugins.callable_with_no_args)
+
+ def test_register(self) -> None:
+ self.assertIsNotNone(self._registry.register('a_plugin',
+ _no_docstring))
+ self.assertIs(self._registry['a_plugin'].target, _no_docstring)
+
+ def test_register_by_name(self) -> None:
+ self.assertIsNotNone(
+ self._registry.register_by_name('plugin_one', __name__,
+ '_no_docstring'))
+ self.assertIsNotNone(
+ self._registry.register('plugin_two', _no_docstring))
+
+ self.assertIs(self._registry['plugin_one'].target, _no_docstring)
+ self.assertIs(self._registry['plugin_two'].target, _no_docstring)
+
+ def test_register_by_name_undefined_module(self) -> None:
+ with self.assertRaisesRegex(plugins.Error, 'No module named'):
+ self._registry.register_by_name('plugin_two', 'not a module',
+ 'something')
+
+ def test_register_by_name_undefined_function(self) -> None:
+ with self.assertRaisesRegex(plugins.Error, 'does not exist'):
+ self._registry.register_by_name('plugin_two', __name__, 'nothing')
+
+ def test_register_fails_validation(self) -> None:
+ with self.assertRaisesRegex(plugins.Error, 'must be callable'):
+ self._registry.register('plugin_two', 'not function')
+
+ def test_run_with_argv_sets_sys_argv(self) -> None:
+ def stash_argv() -> int:
+ self.assertEqual(['pw go', '1', '2'], sys.argv)
+ return 1
+
+ self.assertIsNotNone(self._registry.register('go', stash_argv))
+
+ original_argv = sys.argv
+ self.assertEqual(self._registry.run_with_argv('go', ['1', '2']), 1)
+ self.assertIs(sys.argv, original_argv)
+
+ def test_run_with_argv_registered_plugin(self) -> None:
+ with self.assertRaises(KeyError):
+ self._registry.run_with_argv('plugin_one', [])
+
+ def test_register_cannot_overwrite(self) -> None:
+ self.assertIsNotNone(self._registry.register('foo', lambda: None))
+ self.assertIsNotNone(
+ self._registry.register_by_name('bar', __name__, '_no_docstring'))
+
+ with self.assertRaises(plugins.Error):
+ self._registry.register('foo', lambda: None)
+
+ with self.assertRaises(plugins.Error):
+ self._registry.register('bar', lambda: None)
+
+ def test_register_directory_innermost_takes_priority(self) -> None:
+ with tempfile.TemporaryDirectory() as tempdir:
+ paths = list(_create_files(tempdir, _TEST_PLUGINS))
+ self._registry.register_directory(paths[1].parent, 'TEST_PLUGINS')
+
+ self.assertEqual(self._registry.run_with_argv('test_plugin', []), 123)
+
+ def test_register_directory_only_searches_up(self) -> None:
+ with tempfile.TemporaryDirectory() as tempdir:
+ paths = list(_create_files(tempdir, _TEST_PLUGINS))
+ self._registry.register_directory(paths[0].parent, 'TEST_PLUGINS')
+
+ self.assertEqual(self._registry.run_with_argv('test_plugin', []), 456)
+
+ def test_register_directory_with_restriction(self) -> None:
+ with tempfile.TemporaryDirectory() as tempdir:
+ paths = list(_create_files(tempdir, _TEST_PLUGINS))
+ self._registry.register_directory(paths[0].parent, 'TEST_PLUGINS',
+ Path(tempdir, 'nested', 'in'))
+
+ self.assertNotIn('other_plugin', self._registry)
+
+ def test_register_same_file_multiple_times_no_error(self) -> None:
+ with tempfile.TemporaryDirectory() as tempdir:
+ paths = list(_create_files(tempdir, _TEST_PLUGINS))
+ self._registry.register_file(paths[0])
+ self._registry.register_file(paths[0])
+ self._registry.register_file(paths[0])
+
+ self.assertIs(self._registry['test_plugin'].target, _with_docstring)
+
+ def test_help_uses_function_or_module_docstring(self) -> None:
+ self.assertIsNotNone(self._registry.register('a', _no_docstring))
+ self.assertIsNotNone(self._registry.register('b', _with_docstring))
+
+ self.assertIn(__doc__, '\n'.join(self._registry.detailed_help(['a'])))
+
+ self.assertNotIn(__doc__,
+ '\n'.join(self._registry.detailed_help(['b'])))
+ self.assertIn(_with_docstring.__doc__,
+ '\n'.join(self._registry.detailed_help(['b'])))
+
+ def test_empty_string_if_no_help(self) -> None:
+ fake_module_name = f'{__name__}.fake_module_for_test{id(self)}'
+ fake_module = types.ModuleType(fake_module_name)
+ self.assertIsNone(fake_module.__doc__)
+
+ sys.modules[fake_module_name] = fake_module
+
+ try:
+
+ function = lambda: None
+ function.__module__ = fake_module_name
+ self.assertIsNotNone(self._registry.register('a', function))
+
+ self.assertEqual(self._registry['a'].help(full=False), '')
+ self.assertEqual(self._registry['a'].help(full=True), '')
+ finally:
+ del sys.modules[fake_module_name]
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/pw_cli/py/pw_cli/__main__.py b/pw_cli/py/pw_cli/__main__.py
index 799cbc4..ad6ceeb 100644
--- a/pw_cli/py/pw_cli/__main__.py
+++ b/pw_cli/py/pw_cli/__main__.py
@@ -19,7 +19,7 @@
from typing import NoReturn
import pw_cli.log
-from pw_cli import arguments, plugins
+from pw_cli import arguments, plugins, pw_command_plugins
_LOG = logging.getLogger(__name__)
@@ -38,15 +38,15 @@
_LOG.debug('Executing the pw command from %s', args.directory)
os.chdir(args.directory)
- plugins.register(args.directory)
+ pw_command_plugins.register(args.directory)
if args.help or args.command is None:
- print(arguments.format_help(), file=sys.stderr)
+ print(pw_command_plugins.format_help(), file=sys.stderr)
sys.exit(0)
try:
- sys.exit(plugins.run(args.command, args.plugin_args))
- except plugins.Error as err:
+ sys.exit(pw_command_plugins.run(args.command, args.plugin_args))
+ except (plugins.Error, KeyError) as err:
_LOG.critical('Cannot run command %s.', args.command)
_LOG.critical('%s', err)
sys.exit(2)
diff --git a/pw_cli/py/pw_cli/arguments.py b/pw_cli/py/pw_cli/arguments.py
index 7a0f1b2..f9e2358 100644
--- a/pw_cli/py/pw_cli/arguments.py
+++ b/pw_cli/py/pw_cli/arguments.py
@@ -39,9 +39,9 @@
print(banner(), file=sys.stderr)
-def format_help() -> str:
+def format_help(registry: plugins.Registry) -> str:
"""Returns the pw help information as a string."""
- return f'{_parser().format_help()}\n{plugins.command_help()}'
+ return f'{_parser().format_help()}\n{registry.short_help()}'
class _ArgumentParserWithBanner(argparse.ArgumentParser):
diff --git a/pw_cli/py/pw_cli/plugins.py b/pw_cli/py/pw_cli/plugins.py
index 09ea5d5..540b09c 100644
--- a/pw_cli/py/pw_cli/plugins.py
+++ b/pw_cli/py/pw_cli/plugins.py
@@ -1,4 +1,4 @@
-# Copyright 2020 The Pigweed Authors
+# Copyright 2021 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
@@ -11,28 +11,40 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
-"""Registry for plugins."""
+"""Provides general purpose plugin functionality.
-import argparse
+As used in this module, a plugin is a Python object associated with a name.
+Plugins are registered in a Registry. The plugin object is typically a function,
+but can be anything.
+
+Plugins may be loaded in a variety of ways:
+
+- Listed in a plugins file in the file system (e.g. as "name module target").
+- Registered in a Python file using a decorator (@my_registry.plugin).
+- Registered directly or by name with function calls on a registry object.
+
+This functionality can be used to create plugins for command line tools,
+interactive consoles, or anything else. Pigweed's pw command uses this module
+for its plugins.
+"""
+
import collections
+import collections.abc
import importlib
import inspect
import logging
-import os
from pathlib import Path
import sys
from textwrap import TextWrapper
-from typing import Callable, Dict, List, Iterable, Optional, Union
-
-from pw_cli import arguments
+import types
+from typing import Any, Callable, Dict, List, Iterable, Iterator, Optional, Set
_LOG = logging.getLogger(__name__)
-
-REGISTRY_FILE = 'PW_PLUGINS'
+_BUILT_IN = '<built-in>'
class Error(Exception):
- """Failed to register a Pigweed plugin."""
+ """Indicates that a plugin is invalid or cannot be registered."""
def __str__(self):
"""Displays the error as a string, including the __cause__ if present.
@@ -45,145 +57,312 @@
f'({type(self.__cause__).__name__}: {self.__cause__})')
-class _Plugin:
- """A plugin for the pw command."""
- def __init__(self, name: str, module: str, function: str,
- source: Union[Path, str]):
- self.name = name
- self.source = source
+def _get_module(member: object) -> types.ModuleType:
+ """Gets the module or a dummy module if the module isn't found."""
+ module = inspect.getmodule(member)
+ return module if module else types.ModuleType('<unknown>')
- # Attempt to access the module and function. Catch any errors that might
- # occur, since a bad plugin shouldn't break the rest of the pw command.
+
+class Plugin:
+ """Represents a Python entity registered as a plugin.
+
+ Each plugin resolves to a Python object, typically a function.
+ """
+ @classmethod
+ def from_name(cls, name: str, module_name: str, member_name: str,
+ source: Optional[Path]) -> 'Plugin':
+ """Creates a plugin by module and attribute name.
+
+ Args:
+ name: the name of the plugin
+ module_name: Python module name (e.g. 'foo_pkg.bar')
+ member_name: the name of the member in the module
+ source: path to the plugins file that declared this plugin, if any
+ """
+
+ # Attempt to access the module and member. Catch any errors that might
+ # occur, since a bad plugin shouldn't be a fatal error.
try:
- self._module = importlib.import_module(module)
+ module = importlib.import_module(module_name)
except Exception as err:
- raise Error(f'Failed to import module "{module}"') from err
+ raise Error(f'Failed to import module "{module_name}"') from err
try:
- self._function: Callable[[], int] = getattr(self._module, function)
+ member = getattr(module, member_name)
except AttributeError as err:
raise Error(
- f'The function "{module}.{function}" does not exist') from err
+ f'"{module_name}.{member_name}" does not exist') from err
- try:
- params = inspect.signature(self._function).parameters
- except TypeError:
- raise Error(
- 'Plugin functions must be callable, but '
- f'{module}.{function} is a {type(self._function).__name__}')
+ return cls(name, member, source)
- positional_args = sum(p.default == p.empty for p in params.values())
- if positional_args:
- raise Error(
- f'Plugin functions cannot have any required positional '
- f'arguments, but {module}.{function} has {positional_args}')
+ def __init__(self, name: str, target: Any, source: Path = None) -> None:
+ """Creates a plugin for the provided target."""
+ self.name = name
+ self._module = _get_module(target)
+ self.target = target
+ self.source = source
- def run(self, args: List[str]) -> int:
+ @property
+ def target_name(self) -> str:
+ return (f'{self._module.__name__}.'
+ f'{getattr(self.target, "__name__", self.target)}')
+
+ @property
+ def source_name(self) -> str:
+ return _BUILT_IN if self.source is None else str(self.source)
+
+ def run_with_argv(self, argv: Iterable[str]) -> int:
+ """Sets sys.argv and calls the plugin function.
+
+ This is used to call a plugin as if from the command line.
+ """
original_sys_argv = sys.argv
- sys.argv = [f'pw {self.name}', *args]
+ sys.argv = [f'pw {self.name}', *argv]
try:
- return self._function()
+ return self.target()
finally:
sys.argv = original_sys_argv
- def help(self) -> str:
- """Returns a brief description of this plugin."""
- return self._function.__doc__ or self._module.__doc__ or ''
+ def help(self, full: bool = False) -> str:
+ """Returns a description of this plugin from its docstring."""
+ docstring = self.target.__doc__ or self._module.__doc__ or ''
+ return docstring if full else next(iter(docstring.splitlines()), '')
- def details(self) -> List[str]:
- return [
- f'help {self.help()}',
- f'module {self._module.__name__}',
- f'function {self._function.__name__}',
- f'source {self.source}',
- ]
+ def details(self, full: bool = False) -> Iterator[str]:
+ yield f'help {self.help(full=full)}'
+ yield f'module {self._module.__name__}'
+ yield f'target {getattr(self.target, "__name__", self.target)}'
+ yield f'source {self.source_name}'
+
+ def __repr__(self) -> str:
+ return (f'{self.__class__.__name__}(name={self.name!r}, '
+ f'target={self.target_name}'
+ f'{f", source={self.source_name!r}" if self.source else ""})')
-# This is the global CLI plugin registry.
-_registry: Dict[str, _Plugin] = {}
-_sources: List[Path] = [] # Paths to PW_PLUGINS files
-_errors: Dict[str, List[Error]] = collections.defaultdict(list)
+def callable_with_no_args(plugin: Plugin) -> None:
+ """Checks that a plugin is callable without arguments.
-
-def _get(name: str) -> _Plugin:
- if name in _registry:
- return _registry[name]
-
- if name in _errors:
- raise Error(f'Registration for "{name}" failed: ' +
- ', '.join(str(e) for e in _errors[name]))
-
- raise Error(f'The plugin "{name}" has not been registered')
-
-
-def errors() -> Dict[str, List[Error]]:
- return _errors
-
-
-def run(name: str, args: List[str]) -> int:
- """Runs a plugin by name. Raises Error if the plugin is not registered."""
- return _get(name).run(args)
-
-
-def command_help() -> str:
- """Returns a help string for the registered plugins."""
- width = max(len(name) for name in _registry) + 1 if _registry else 1
- help_items = '\n'.join(f' {name:{width}} {plugin.help()}'
- for name, plugin in sorted(_registry.items()))
- return f'supported commands:\n{help_items}'
-
-
-_BUILTIN_PLUGIN = '<built-in>'
-
-
-def _valid_registration(name: str, module: str, function: str,
- source: Union[Path, str]) -> bool:
- """Determines if a plugin should be registered or not."""
- existing = _registry.get(name)
-
- if existing is None:
- return True
-
- if source == _BUILTIN_PLUGIN:
- raise Error(
- f'Attempted to register built-in plugin "{name}", but that '
- f'plugin was previously registered ({existing.source})!')
-
- if existing.source == _BUILTIN_PLUGIN:
- _LOG.debug('%s: Overriding built-in plugin "%s" with %s.%s', source,
- name, module, function)
- return True
-
- if source == _registry[name].source:
- _LOG.warning(
- '%s: "%s" is registered multiple times in this file! '
- 'Only the first registration takes effect', source, name)
- else:
- _LOG.debug(
- '%s: The plugin "%s" was previously registered in %s; '
- 'ignoring registration as %s.%s', source, name,
- _registry[name].source, module, function)
-
- return False
-
-
-def _register(name: str,
- module: str,
- function: str,
- source: Union[Path, str] = _BUILTIN_PLUGIN) -> None:
- """Registers a plugin from the specified source."""
-
- if not _valid_registration(name, module, function, source):
- return
-
+ May be used for the validator argument to Registry.
+ """
try:
- _registry[name] = _Plugin(name, module, function, source)
- _LOG.debug('%s: Registered plugin "%s" for %s.%s', source, name,
- module, function)
- except Error as err:
- _errors[name].append(err)
- _LOG.error('%s: Failed to register plugin "%s": %s', source, name, err)
+ params = inspect.signature(plugin.target).parameters
+ except TypeError:
+ raise Error('Plugin functions must be callable, but '
+ f'{plugin.target_name} is a '
+ f'{type(plugin.target).__name__}')
+
+ positional = sum(p.default == p.empty for p in params.values())
+ if positional:
+ raise Error(f'Plugin functions cannot have any required positional '
+ f'arguments, but {plugin.target_name} has {positional}')
+
+
+class Registry(collections.abc.Mapping):
+ """Manages a set of plugins from Python modules or plugins files."""
+ def __init__(self,
+ validator: Callable[[Plugin], Any] = lambda _: None) -> None:
+ """Creates a new, empty plugins registry.
+
+ Args:
+ validator: Function that checks whether a plugin is valid and should
+ be registered. Must raise plugins.Error is the plugin is invalid.
+ """
+
+ self._registry: Dict[str, Plugin] = {}
+ self._sources: Set[Path] = set() # Paths to plugins files
+ self._errors: Dict[str,
+ List[Exception]] = collections.defaultdict(list)
+ self._validate_plugin = validator
+
+ def __getitem__(self, name: str) -> Plugin:
+ """Accesses a plugin by name; raises KeyError if it does not exist."""
+ if name in self._registry:
+ return self._registry[name]
+
+ if name in self._errors:
+ raise KeyError(f'Registration for "{name}" failed: ' +
+ ', '.join(str(e) for e in self._errors[name]))
+
+ raise KeyError(f'The plugin "{name}" has not been registered')
+
+ def __iter__(self) -> Iterator[str]:
+ return iter(self._registry)
+
+ def __len__(self) -> int:
+ return len(self._registry)
+
+ def errors(self) -> Dict[str, List[Exception]]:
+ return self._errors
+
+ def run_with_argv(self, name: str, argv: Iterable[str]) -> int:
+ """Runs a plugin by name, setting sys.argv to the provided args.
+
+ This is used to run a command as if it were executed directly from the
+ command line. The plugin is expected to return an int.
+
+ Raises:
+ KeyError if plugin is not registered.
+ """
+ return self[name].run_with_argv(argv)
+
+ def _should_register(self, plugin: Plugin) -> bool:
+ """Determines and logs if a plugin should be registered or not.
+
+ Some errors are exceptions, others are not.
+ """
+
+ if plugin.name in self._registry and plugin.source is None:
+ raise Error(
+ f'Attempted to register built-in plugin "{plugin.name}", but '
+ 'a plugin with that name was previously registered '
+ f'({self[plugin.name]})!')
+
+ # Run the user-provided validation function, which raises exceptions
+ # if there are errors.
+ self._validate_plugin(plugin)
+
+ existing = self._registry.get(plugin.name)
+
+ if existing is None:
+ return True
+
+ if existing.source is None:
+ _LOG.debug('%s: Overriding built-in plugin "%s" with %s',
+ plugin.source_name, plugin.name, plugin.target_name)
+ return True
+
+ if plugin.source != existing.source:
+ _LOG.debug(
+ '%s: The plugin "%s" was previously registered in %s; '
+ 'ignoring registration as %s', plugin.source_name, plugin.name,
+ self._registry[plugin.name].source, plugin.target_name)
+ elif plugin.source not in self._sources:
+ _LOG.warning(
+ '%s: "%s" is registered file multiple times in this file! '
+ 'Only the first registration takes effect', plugin.source_name,
+ plugin.name)
+
+ return False
+
+ def register(self, name: str, target: Any) -> Optional[Plugin]:
+ """Registers an object as a plugin."""
+ return self._register(Plugin(name, target, None))
+
+ def register_by_name(self,
+ name: str,
+ module_name: str,
+ member_name: str,
+ source: Path = None) -> Optional[Plugin]:
+ """Registers an object from its module and name as a plugin."""
+ return self._register(
+ Plugin.from_name(name, module_name, member_name, source))
+
+ def _register(self, plugin: Plugin) -> Optional[Plugin]:
+ # Prohibit functions not from a plugins file from overriding others.
+ if not self._should_register(plugin):
+ return None
+
+ self._registry[plugin.name] = plugin
+ _LOG.debug('%s: Registered plugin "%s" for %s', plugin.source_name,
+ plugin.name, plugin.target_name)
+
+ return plugin
+
+ def register_file(self, path: Path) -> None:
+ """Registers plugins from a plugins file.
+
+ Any exceptions raised from parsing the file are caught and logged.
+ """
+ with path.open() as contents:
+ for lineno, line in enumerate(contents, 1):
+ line = line.strip()
+ if not line or line.startswith('#'):
+ continue
+
+ try:
+ name, module, function = line.split()
+ except ValueError as err:
+ self._errors[line.strip()].append(Error(err))
+ _LOG.error(
+ '%s:%d: Failed to parse plugin entry "%s": '
+ 'Expected 3 items (name, module, function), '
+ 'got %d', path, lineno, line, len(line.split()))
+ continue
+
+ try:
+ self.register_by_name(name, module, function, path)
+ except Error as err:
+ self._errors[name].append(err)
+ _LOG.error('%s: Failed to register plugin "%s": %s', path,
+ name, err)
+
+ self._sources.add(path)
+
+ def register_directory(self,
+ directory: Path,
+ file_name: str,
+ restrict_to: Path = None) -> None:
+ """Finds and registers plugins from plugins files in a directory.
+
+ Args:
+ directory: The directory from which to start searching up.
+ file_name: The name of plugins files to look for.
+ restrict_to: If provided, do not search higher than this directory.
+ """
+ for path in find_all_in_parents(file_name, directory):
+ if not path.is_file():
+ continue
+
+ if restrict_to is not None and restrict_to not in path.parents:
+ _LOG.debug(
+ "Skipping plugins file %s because it's outside of %s",
+ path, restrict_to)
+ continue
+
+ _LOG.debug('Found plugins file %s', path)
+ self.register_file(path)
+
+ def short_help(self) -> str:
+ """Returns a help string for the registered plugins."""
+ width = max(len(name)
+ for name in self._registry) + 1 if self._registry else 1
+ help_items = '\n'.join(
+ f' {name:{width}} {plugin.help()}'
+ for name, plugin in sorted(self._registry.items()))
+ return f'supported plugins:\n{help_items}'
+
+ def detailed_help(self, plugins: Iterable[str] = ()) -> Iterator[str]:
+ """Yields lines of detailed information about commands."""
+ if not plugins:
+ plugins = list(self._registry)
+
+ yield '\ndetailed plugin information:'
+
+ wrapper = TextWrapper(width=80,
+ initial_indent=' ',
+ subsequent_indent=' ' * 11)
+
+ plugins = sorted(plugins)
+ for plugin in plugins:
+ yield f' [{plugin}]'
+
+ try:
+ for line in self[plugin].details(full=len(plugins) == 1):
+ yield wrapper.fill(line)
+ except KeyError as err:
+ yield wrapper.fill(f'error {str(err)[1:-1]}')
+
+ yield ''
+
+ yield 'Plugins files:'
+
+ if self._sources:
+ yield from (f' [{i}] {file}'
+ for i, file in enumerate(self._sources, 1))
+ else:
+ yield ' (none found)'
def find_in_parents(name: str, path: Path) -> Optional[Path]:
@@ -199,7 +378,7 @@
return path.joinpath(name)
-def find_all_in_parents(name: str, path: Path) -> Iterable[Path]:
+def find_all_in_parents(name: str, path: Path) -> Iterator[Path]:
"""Searches all parent directories of the path for files or directories."""
while True:
@@ -209,93 +388,3 @@
yield result
path = result.parent.parent
-
-
-def _register_builtin_plugins():
- """Registers the commands that are included with pw by default."""
-
- _register('doctor', 'pw_doctor.doctor', 'main')
- _register('format', 'pw_presubmit.format_code', 'main')
- _register('help', 'pw_cli.plugins', '_help_command')
- _register('logdemo', 'pw_cli.log', 'main')
- _register('module-check', 'pw_module.check', 'main')
- _register('test', 'pw_unit_test.test_runner', 'main')
- _register('watch', 'pw_watch.watch', 'main')
-
-
-def register(directory: Path):
- """Finds and registers command line plugins."""
- _register_builtin_plugins()
-
- # Find pw plugins files starting in the current and parent directories.
- for path in find_all_in_parents(REGISTRY_FILE, directory):
- if not path.is_file():
- continue
-
- root = Path(os.environ.get('PW_PROJECT_ROOT', '')).resolve()
- if root not in path.parents:
- _LOG.debug(
- "Skipping plugins file %s because it's outside of "
- 'PW_PROJECT_ROOT (%s)', path, root)
- continue
-
- _LOG.debug('Found plugins file %s', path)
- _sources.append(path)
-
- with path.open() as contents:
- for lineno, line in enumerate(contents, 1):
- line = line.strip()
- if line and not line.startswith('#'):
- try:
- name, module, function = line.split()
- _register(name, module, function, path)
- except ValueError as err:
- _errors[line.strip()].append(Error(err))
- _LOG.error(
- '%s:%d: Failed to parse plugin entry "%s": '
- 'Expected 3 items (name, module, function), got %d',
- path, lineno, line, len(line.split()))
-
-
-def _help_text(plugins: Iterable[str] = ()) -> Iterable[str]:
- """Yields detailed information about commands."""
- yield arguments.format_help()
-
- if not plugins:
- plugins = list(_registry)
-
- yield '\ndetailed command information:'
-
- wrapper = TextWrapper(width=80,
- initial_indent=' ',
- subsequent_indent=' ' * 13)
-
- for plugin in sorted(plugins):
- yield f' [{plugin}]'
-
- try:
- for line in _get(plugin).details():
- yield wrapper.fill(line)
- except Error as err:
- yield wrapper.fill(f'error {err}')
-
- yield ''
-
- yield 'PW_PLUGINS files:'
-
- if _sources:
- yield from (f' [{i}] {file}' for i, file in enumerate(_sources, 1))
- else:
- yield ' (none found)'
-
-
-def _help_command():
- """Display detailed information about pw commands."""
- parser = argparse.ArgumentParser(description=_help_command.__doc__)
- parser.add_argument('plugins',
- metavar='plugin',
- nargs='*',
- help='command for which to display detailed info')
-
- for line in _help_text(**vars(parser.parse_args())):
- print(line, file=sys.stderr)
diff --git a/pw_cli/py/pw_cli/pw_command_plugins.py b/pw_cli/py/pw_cli/pw_command_plugins.py
new file mode 100644
index 0000000..a5c20df
--- /dev/null
+++ b/pw_cli/py/pw_cli/pw_command_plugins.py
@@ -0,0 +1,71 @@
+# Copyright 2021 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""This module manages the global plugin registry for pw_cli."""
+
+import argparse
+import os
+from pathlib import Path
+import sys
+from typing import Iterable
+
+from pw_cli import arguments, plugins
+
+_plugin_registry = plugins.Registry(validator=plugins.callable_with_no_args)
+REGISTRY_FILE = 'PW_PLUGINS'
+
+
+def _register_builtin_plugins(registry: plugins.Registry) -> None:
+ """Registers the commands that are included with pw by default."""
+
+ # Register these by name to avoid circular dependencies.
+ registry.register_by_name('doctor', 'pw_doctor.doctor', 'main')
+ registry.register_by_name('format', 'pw_presubmit.format_code', 'main')
+ registry.register_by_name('logdemo', 'pw_cli.log', 'main')
+ registry.register_by_name('module-check', 'pw_module.check', 'main')
+ registry.register_by_name('test', 'pw_unit_test.test_runner', 'main')
+ registry.register_by_name('watch', 'pw_watch.watch', 'main')
+
+ registry.register('help', _help_command)
+
+
+def _help_command():
+ """Display detailed information about pw commands."""
+ parser = argparse.ArgumentParser(description=_help_command.__doc__)
+ parser.add_argument('plugins',
+ metavar='plugin',
+ nargs='*',
+ help='command for which to display detailed info')
+
+ print(arguments.format_help(_plugin_registry), file=sys.stderr)
+
+ for line in _plugin_registry.detailed_help(**vars(parser.parse_args())):
+ print(line, file=sys.stderr)
+
+
+def register(directory: Path) -> None:
+ _register_builtin_plugins(_plugin_registry)
+ _plugin_registry.register_directory(
+ directory, REGISTRY_FILE, Path(os.environ.get('PW_PROJECT_ROOT', '')))
+
+
+def errors() -> dict:
+ return _plugin_registry.errors()
+
+
+def format_help() -> str:
+ return arguments.format_help(_plugin_registry)
+
+
+def run(name: str, args: Iterable[str]) -> int:
+ return _plugin_registry.run_with_argv(name, args)
diff --git a/pw_doctor/py/pw_doctor/doctor.py b/pw_doctor/py/pw_doctor/doctor.py
index 16fb220..7594616 100755
--- a/pw_doctor/py/pw_doctor/doctor.py
+++ b/pw_doctor/py/pw_doctor/doctor.py
@@ -26,7 +26,7 @@
import tempfile
from typing import Callable, Iterable, List, Set
-import pw_cli.plugins
+import pw_cli.pw_command_plugins
def call_stdout(*args, **kwargs):
@@ -130,7 +130,7 @@
@register_into(CHECKS)
def pw_plugins(ctx: DoctorContext):
- if pw_cli.plugins.errors():
+ if pw_cli.pw_command_plugins.errors():
ctx.error('Not all pw plugins loaded successfully')