diff --git a/docs/dev-guide.rst b/docs/dev-guide.rst index 8591987a..16b6c8ae 100644 --- a/docs/dev-guide.rst +++ b/docs/dev-guide.rst @@ -122,6 +122,42 @@ Also notice plugins are activated in a specific order, using Python's built-in ``sorted`` function. +Providing multiple schemas +-------------------------- + +A second system is provided for providing multiple schemas in a single plugin. +This is useful when a single plugin is responsible for multiple subtables +under the ``tool`` table, or if you need to provide multiple schemas for a +a single subtable. + +To use this system, the plugin function, which does not take any arguments, +should return a dictionary with two keys: ``tools``, which is a dictionary of +tool names to schemas, and optionally ``schemas``, which is a list of schemas +that are not associated with any specific tool, but are loaded via ref's from +the other tools. + +When using a :pep:`621`-compliant backend, the following can be add to your +``pyproject.toml`` file: + +.. code-block:: toml + + # in pyproject.toml + [project.entry-points."validate_pyproject.validate_pyproject.multi_schema"] + arbitrary = "your_package.your_module:your_plugin" + +An example of the plugin structure needed for this system is shown below: + +.. code-block:: python + + def your_plugin(tool_name: str) -> dict: + return { + "tools": {"my-tool": my_schema}, + "schemas": [my_extra_schema], + } + +Fragments for schemas are also supported with this system; use ``#`` to split +the tool name and fragment path in the dictionary key. + .. _entry-point: https://setuptools.pypa.io/en/stable/userguide/entry_point.html#entry-points .. _JSON Schema: https://json-schema.org/ .. _Python package: https://packaging.python.org/ diff --git a/src/validate_pyproject/cli.py b/src/validate_pyproject/cli.py index 37a59713..a356640c 100644 --- a/src/validate_pyproject/cli.py +++ b/src/validate_pyproject/cli.py @@ -30,7 +30,7 @@ from . import _tomllib as tomllib from .api import Validator from .errors import ValidationError -from .plugins import PluginWrapper +from .plugins import PluginProtocol, PluginWrapper from .plugins import list_from_entry_points as list_plugins_from_entry_points from .remote import RemotePlugin, load_store @@ -124,7 +124,7 @@ class CliParams(NamedTuple): dump_json: bool = False -def __meta__(plugins: Sequence[PluginWrapper]) -> Dict[str, dict]: +def __meta__(plugins: Sequence[PluginProtocol]) -> Dict[str, dict]: """'Hyper parameters' to instruct :mod:`argparse` how to create the CLI""" meta = {k: v.copy() for k, v in META.items()} meta["enable"]["choices"] = {p.tool for p in plugins} @@ -135,9 +135,9 @@ def __meta__(plugins: Sequence[PluginWrapper]) -> Dict[str, dict]: @critical_logging() def parse_args( args: Sequence[str], - plugins: Sequence[PluginWrapper], + plugins: Sequence[PluginProtocol], description: str = "Validate a given TOML file", - get_parser_spec: Callable[[Sequence[PluginWrapper]], Dict[str, dict]] = __meta__, + get_parser_spec: Callable[[Sequence[PluginProtocol]], Dict[str, dict]] = __meta__, params_class: Type[T] = CliParams, # type: ignore[assignment] ) -> T: """Parse command line parameters @@ -167,11 +167,14 @@ def parse_args( return params_class(**params) # type: ignore[call-overload, no-any-return] +Plugins = TypeVar("Plugins", bound=PluginProtocol) + + def select_plugins( - plugins: Sequence[PluginWrapper], + plugins: Sequence[Plugins], enabled: Sequence[str] = (), disabled: Sequence[str] = (), -) -> List[PluginWrapper]: +) -> List[Plugins]: available = list(plugins) if enabled: available = [p for p in available if p.tool in enabled] @@ -219,7 +222,7 @@ def run(args: Sequence[str] = ()) -> int: (for example ``["--verbose", "setup.cfg"]``). """ args = args or sys.argv[1:] - plugins: List[PluginWrapper] = list_plugins_from_entry_points() + plugins = list_plugins_from_entry_points() params: CliParams = parse_args(args, plugins) setup_logging(params.loglevel) tool_plugins = [RemotePlugin.from_str(t) for t in params.tool] @@ -263,7 +266,7 @@ def _split_lines(self, text: str, width: int) -> List[str]: return list(chain.from_iterable(wrap(x, width) for x in text.splitlines())) -def plugins_help(plugins: Sequence[PluginWrapper]) -> str: +def plugins_help(plugins: Sequence[PluginProtocol]) -> str: return "\n".join(_format_plugin_help(p) for p in plugins) @@ -273,7 +276,7 @@ def _flatten_str(text: str) -> str: return (text[0].lower() + text[1:]).strip() -def _format_plugin_help(plugin: PluginWrapper) -> str: +def _format_plugin_help(plugin: PluginProtocol) -> str: help_text = plugin.help_text help_text = f": {_flatten_str(help_text)}" if help_text else "" return f"* {plugin.tool!r}{help_text}" diff --git a/src/validate_pyproject/plugins/__init__.py b/src/validate_pyproject/plugins/__init__.py index 19ca2c14..cb1dd8dd 100644 --- a/src/validate_pyproject/plugins/__init__.py +++ b/src/validate_pyproject/plugins/__init__.py @@ -7,15 +7,24 @@ import typing from importlib.metadata import EntryPoint, entry_points +from itertools import chain from string import Template from textwrap import dedent -from typing import Any, Callable, Iterable, List, Optional, Protocol +from typing import ( + Any, + Callable, + Generator, + Iterable, + List, + NamedTuple, + Optional, + Protocol, + Union, +) from .. import __version__ from ..types import Plugin, Schema -ENTRYPOINT_GROUP = "validate_pyproject.tool_schema" - class PluginProtocol(Protocol): @property @@ -66,34 +75,63 @@ def __repr__(self) -> str: return f"{self.__class__.__name__}({self.tool!r}, {self.id})" +class StoredPlugin: + def __init__(self, tool: str, schema: Schema): + self._tool, _, self._fragment = tool.partition("#") + self._schema = schema + + @property + def id(self) -> str: + return self.schema.get("id", "MISSING ID") + + @property + def tool(self) -> str: + return self._tool + + @property + def schema(self) -> Schema: + return self._schema + + @property + def fragment(self) -> str: + return self._fragment + + @property + def help_text(self) -> str: + return self.schema.get("description", "") + + def __repr__(self) -> str: + args = [repr(self.tool), self.id] + if self.fragment: + args.append(f"fragment={self.fragment!r}") + return f"{self.__class__.__name__}({', '.join(args)}, )" + + if typing.TYPE_CHECKING: _: PluginProtocol = typing.cast(PluginWrapper, None) -def iterate_entry_points(group: str = ENTRYPOINT_GROUP) -> Iterable[EntryPoint]: - """Produces a generator yielding an EntryPoint object for each plugin registered +def iterate_entry_points(group: str) -> Iterable[EntryPoint]: + """Produces an iterable yielding an EntryPoint object for each plugin registered via ``setuptools`` `entry point`_ mechanism. This method can be used in conjunction with :obj:`load_from_entry_point` to filter - the plugins before actually loading them. + the plugins before actually loading them. The entry points are not + deduplicated. """ entries = entry_points() if hasattr(entries, "select"): # pragma: no cover # The select method was introduced in importlib_metadata 3.9 (and Python 3.10) # and the previous dict interface was declared deprecated select = typing.cast( - Any, + Callable[..., Iterable[EntryPoint]], getattr(entries, "select"), # noqa: B009 ) # typecheck gymnastics - entries_: Iterable[EntryPoint] = select(group=group) - else: # pragma: no cover - # TODO: Once Python 3.10 becomes the oldest version supported, this fallback and - # conditional statement can be removed. - entries_ = (plugin for plugin in entries.get(group, [])) - deduplicated = { - e.name: e for e in sorted(entries_, key=lambda e: (e.name, e.value)) - } - return list(deduplicated.values()) + return select(group=group) + # pragma: no cover + # TODO: Once Python 3.10 becomes the oldest version supported, this fallback and + # conditional statement can be removed. + return (plugin for plugin in entries.get(group, [])) def load_from_entry_point(entry_point: EntryPoint) -> PluginWrapper: @@ -105,23 +143,64 @@ def load_from_entry_point(entry_point: EntryPoint) -> PluginWrapper: raise ErrorLoadingPlugin(entry_point=entry_point) from ex +def load_from_multi_entry_point( + entry_point: EntryPoint, +) -> Generator[StoredPlugin, None, None]: + """Carefully load the plugin, raising a meaningful message in case of errors""" + try: + fn = entry_point.load() + output = fn() + except Exception as ex: + raise ErrorLoadingPlugin(entry_point=entry_point) from ex + + for tool, schema in output["tools"].items(): + yield StoredPlugin(tool, schema) + for schema in output.get("schemas", []): + yield StoredPlugin("", schema) + + +class _SortablePlugin(NamedTuple): + priority: int + name: str + plugin: Union[PluginWrapper, StoredPlugin] + + def __lt__(self, other: Any) -> bool: + return (self.plugin.tool or self.plugin.id, self.name, self.priority) < ( + other.plugin.tool or other.plugin.id, + other.name, + other.priority, + ) + + def list_from_entry_points( - group: str = ENTRYPOINT_GROUP, filtering: Callable[[EntryPoint], bool] = lambda _: True, -) -> List[PluginWrapper]: +) -> List[Union[PluginWrapper, StoredPlugin]]: """Produces a list of plugin objects for each plugin registered via ``setuptools`` `entry point`_ mechanism. Args: - group: name of the setuptools' entry point group where plugins is being - registered filtering: function returning a boolean deciding if the entry point should be loaded and included (or not) in the final list. A ``True`` return means the plugin should be included. """ - return [ - load_from_entry_point(e) for e in iterate_entry_points(group) if filtering(e) - ] + tool_eps = ( + _SortablePlugin(0, e.name, load_from_entry_point(e)) + for e in iterate_entry_points("validate_pyproject.tool_schema") + if filtering(e) + ) + multi_eps = ( + _SortablePlugin(1, e.name, p) + for e in sorted( + iterate_entry_points("validate_pyproject.multi_schema"), + key=lambda e: e.name, + reverse=True, + ) + for p in load_from_multi_entry_point(e) + if filtering(e) + ) + eps = chain(tool_eps, multi_eps) + dedup = {e.plugin.tool or e.plugin.id: e.plugin for e in sorted(eps, reverse=True)} + return list(dedup.values())[::-1] class ErrorLoadingPlugin(RuntimeError): diff --git a/src/validate_pyproject/pre_compile/cli.py b/src/validate_pyproject/pre_compile/cli.py index 5fb157cb..46e538e4 100644 --- a/src/validate_pyproject/pre_compile/cli.py +++ b/src/validate_pyproject/pre_compile/cli.py @@ -85,7 +85,9 @@ class CliParams(NamedTuple): store: str = "" -def parser_spec(plugins: Sequence[PluginWrapper]) -> Dict[str, dict]: +def parser_spec( + plugins: Sequence[PluginProtocol], +) -> Dict[str, dict]: common = ("version", "enable", "disable", "verbose", "very_verbose") cli_spec = cli.__meta__(plugins) meta = {k: v.copy() for k, v in META.items()} diff --git a/src/validate_pyproject/repo_review.py b/src/validate_pyproject/repo_review.py index 09fc779d..51c93a6c 100644 --- a/src/validate_pyproject/repo_review.py +++ b/src/validate_pyproject/repo_review.py @@ -28,9 +28,9 @@ def repo_review_checks() -> Dict[str, VPP001]: def repo_review_families(pyproject: Dict[str, Any]) -> Dict[str, Dict[str, str]]: has_distutils = "distutils" in pyproject.get("tool", {}) - plugin_names = (ep.name for ep in plugins.iterate_entry_points()) - plugin_list = ( - f"`[tool.{n}]`" for n in plugin_names if n != "distutils" or has_distutils + plugin_list = plugins.list_from_entry_points( + lambda e: e.name != "distutils" or has_distutils ) - descr = f"Checks `[build-system]`, `[project]`, {', '.join(plugin_list)}" + plugin_names = (f"`[tool.{n.tool}]`" for n in plugin_list if n.tool) + descr = f"Checks `[build-system]`, `[project]`, {', '.join(plugin_names)}" return {"validate-pyproject": {"name": "Validate-PyProject", "description": descr}} diff --git a/tests/test_plugins.py b/tests/test_plugins.py index 1f07dec5..89f3c8bd 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -1,12 +1,16 @@ # The code in this module is mostly borrowed/adapted from PyScaffold and was originally # published under the MIT license # The original PyScaffold license can be found in 'NOTICE.txt' -from importlib.metadata import EntryPoint # pragma: no cover +import functools +import importlib.metadata +import sys +from types import ModuleType +from typing import List import pytest from validate_pyproject import plugins -from validate_pyproject.plugins import ENTRYPOINT_GROUP, ErrorLoadingPlugin +from validate_pyproject.plugins import ErrorLoadingPlugin, PluginWrapper, StoredPlugin EXISTING = ( "setuptools", @@ -18,7 +22,9 @@ def test_load_from_entry_point__error(): # This module does not exist, so Python will have some trouble loading it # EntryPoint(name, value, group) entry = "mypkg.SOOOOO___fake___:activate" - fake = EntryPoint("fake", entry, ENTRYPOINT_GROUP) + fake = importlib.metadata.EntryPoint( + "fake", entry, "validate_pyproject.tool_schema" + ) with pytest.raises(ErrorLoadingPlugin): plugins.load_from_entry_point(fake) @@ -28,7 +34,7 @@ def is_entry_point(ep): def test_iterate_entry_points(): - plugin_iter = plugins.iterate_entry_points() + plugin_iter = plugins.iterate_entry_points("validate_pyproject.tool_schema") assert hasattr(plugin_iter, "__iter__") pluging_list = list(plugin_iter) assert all(is_entry_point(e) for e in pluging_list) @@ -68,3 +74,155 @@ def _fn2(_): pw = plugins.PluginWrapper("name", _fn2) assert pw.help_text == "Help for `name`" + + +class TestStoredPlugin: + def test_empty_help_text(self): + def _fn1(_): + return {} + + pw = plugins.StoredPlugin("name", {}) + assert pw.help_text == "" + + def _fn2(_): + """Help for `${tool}`""" + return {} + + pw = plugins.StoredPlugin("name", {"description": "Help for me"}) + assert pw.help_text == "Help for me" + + +def fake_multi_iterate_entry_points(name: str) -> List[importlib.metadata.EntryPoint]: + if name == "validate_pyproject.multi_schema": + return [ + importlib.metadata.EntryPoint( + name="_", value="test_module:f", group="validate_pyproject.multi_schema" + ) + ] + return [] + + +def test_multi_plugins(monkeypatch): + s1 = {"id": "example1"} + s2 = {"id": "example2"} + s3 = {"id": "example3"} + sys.modules["test_module"] = ModuleType("test_module") + sys.modules["test_module"].f = lambda: { + "tools": {"example#frag": s1}, + "schemas": [s2, s3], + } # type: ignore[attr-defined] + monkeypatch.setattr( + plugins, "iterate_entry_points", fake_multi_iterate_entry_points + ) + + lst = plugins.list_from_entry_points() + assert len(lst) == 3 + assert all(e.id.startswith("example") for e in lst) + + (fragmented,) = (e for e in lst if e.tool) + assert fragmented.tool == "example" + assert fragmented.fragment == "frag" + assert fragmented.schema == s1 + + +def fake_both_iterate_entry_points(name: str) -> List[importlib.metadata.EntryPoint]: + if name == "validate_pyproject.multi_schema": + return [ + importlib.metadata.EntryPoint( + name="_", value="test_module:f", group="validate_pyproject.multi_schema" + ) + ] + if name == "validate_pyproject.tool_schema": + return [ + importlib.metadata.EntryPoint( + name="example1", + value="test_module:f1", + group="validate_pyproject.tool_schema", + ), + importlib.metadata.EntryPoint( + name="example3", + value="test_module:f3", + group="validate_pyproject.tool_schema", + ), + ] + return [] + + +def test_combined_plugins(monkeypatch): + s1 = {"id": "example1"} + s2 = {"id": "example2"} + sys.modules["test_module"] = ModuleType("test_module") + sys.modules["test_module"].f = lambda: { + "tools": {"example1": s1, "example2": s2}, + } # type: ignore[attr-defined] + sys.modules["test_module"].f1 = lambda _: {"id": "tool1"} # type: ignore[attr-defined] + sys.modules["test_module"].f3 = lambda _: {"id": "tool3"} # type: ignore[attr-defined] + monkeypatch.setattr(plugins, "iterate_entry_points", fake_both_iterate_entry_points) + + lst = plugins.list_from_entry_points() + assert len(lst) == 3 + + assert lst[0].tool == "example1" + assert isinstance(lst[0], StoredPlugin) + + assert lst[1].tool == "example2" + assert isinstance(lst[1], StoredPlugin) + + assert lst[2].tool == "example3" + assert isinstance(lst[2], PluginWrapper) + + +def fake_several_entry_points( + name: str, *, reverse: bool +) -> List[importlib.metadata.EntryPoint]: + if name == "validate_pyproject.multi_schema": + items = [ + importlib.metadata.EntryPoint( + name="a", + value="test_module:f1", + group="validate_pyproject.multi_schema", + ), + importlib.metadata.EntryPoint( + name="b", + value="test_module:f2", + group="validate_pyproject.multi_schema", + ), + ] + return items[::-1] if reverse else items + return [] + + +@pytest.mark.parametrize("reverse", [True, False]) +def test_several_multi_plugins(monkeypatch, reverse): + s1 = {"id": "example1"} + s2 = {"id": "example2"} + s3 = {"id": "example3"} + sys.modules["test_module"] = ModuleType("test_module") + sys.modules["test_module"].f1 = lambda: { + "tools": {"example": s1}, + } # type: ignore[attr-defined] + sys.modules["test_module"].f2 = lambda: { + "tools": {"example": s2, "other": s3}, + } # type: ignore[attr-defined] + monkeypatch.setattr( + plugins, + "iterate_entry_points", + functools.partial(fake_several_entry_points, reverse=reverse), + ) + + (plugin1, plugin2) = plugins.list_from_entry_points() + assert plugin1.id == "example1" + assert plugin2.id == "example3" + + +def test_broken_multi_plugin(monkeypatch): + def broken_ep(): + raise RuntimeError("Broken") + + sys.modules["test_module"] = ModuleType("test_module") + sys.modules["test_module"].f = broken_ep + monkeypatch.setattr( + plugins, "iterate_entry_points", fake_multi_iterate_entry_points + ) + with pytest.raises(ErrorLoadingPlugin): + plugins.list_from_entry_points()