From 9e3c48c0f191bbbef495d031960298745d94dbd1 Mon Sep 17 00:00:00 2001 From: Juliya Smith Date: Mon, 2 Oct 2023 12:04:32 -0500 Subject: [PATCH] fix: plugins issue --- src/ape/plugins/__init__.py | 45 +++-- src/ape_plugins/_cli.py | 164 ++++++---------- src/ape_plugins/utils.py | 268 ++++++++++++++++++++++---- tests/functional/test_plugins.py | 136 ++++++++++--- tests/integration/cli/conftest.py | 2 +- tests/integration/cli/test_plugins.py | 6 +- 6 files changed, 426 insertions(+), 195 deletions(-) diff --git a/src/ape/plugins/__init__.py b/src/ape/plugins/__init__.py index 2b6a61f363..0f5574b2b8 100644 --- a/src/ape/plugins/__init__.py +++ b/src/ape/plugins/__init__.py @@ -2,7 +2,7 @@ import importlib import pkgutil import subprocess -from typing import Any, Callable, Generator, Iterator, List, Optional, Tuple, Type +from typing import Any, Callable, Generator, Iterator, List, Optional, Set, Tuple, Type from ape.__modules__ import __modules__ from ape.exceptions import ApeAttributeError @@ -123,25 +123,6 @@ class PluginManager: def __init__(self) -> None: self.__registered = False - @functools.cached_property - def _plugin_modules(self) -> Tuple[str, ...]: - # NOTE: Unable to use pkgutil.iter_modules() for installed plugins - # because it does not work with editable installs. - # See https://github.com/python/cpython/issues/99805. - result = subprocess.check_output( - ["pip", "list", "--format", "freeze", "--disable-pip-version-check"] - ) - packages = result.decode("utf8").splitlines() - installed_plugin_module_names = { - p.split("==")[0].replace("-", "_") for p in packages if p.startswith("ape-") - } - core_plugin_module_names = { - n for _, n, ispkg in pkgutil.iter_modules() if n.startswith("ape_") - } - - # NOTE: Returns tuple because this shouldn't change. - return tuple(installed_plugin_module_names.union(core_plugin_module_names)) - def __repr__(self): return f"<{self.__class__.__name__}>" @@ -173,6 +154,30 @@ def get_plugin_name_and_hookfn(h): if validated_plugin: yield validated_plugin + @property + def registered_plugins(self) -> Set[str]: + self._register_plugins() + return {x[0] for x in plugin_manager.list_name_plugin()} + + @functools.cached_property + def _plugin_modules(self) -> Tuple[str, ...]: + # NOTE: Unable to use pkgutil.iter_modules() for installed plugins + # because it does not work with editable installs. + # See https://github.com/python/cpython/issues/99805. + result = subprocess.check_output( + ["pip", "list", "--format", "freeze", "--disable-pip-version-check"] + ) + packages = result.decode("utf8").splitlines() + installed_plugin_module_names = { + p.split("==")[0].replace("-", "_") for p in packages if p.startswith("ape-") + } + core_plugin_module_names = { + n for _, n, ispkg in pkgutil.iter_modules() if n.startswith("ape_") + } + + # NOTE: Returns tuple because this shouldn't change. + return tuple(installed_plugin_module_names.union(core_plugin_module_names)) + def _register_plugins(self): if self.__registered: return diff --git a/src/ape_plugins/_cli.py b/src/ape_plugins/_cli.py index 07cbe8f8bd..d26f781de9 100644 --- a/src/ape_plugins/_cli.py +++ b/src/ape_plugins/_cli.py @@ -1,15 +1,19 @@ import subprocess import sys from pathlib import Path -from typing import Collection, Dict, List, Set, Tuple +from typing import List, Tuple import click from ape.cli import ape_cli_context, skip_confirmation_option from ape.managers.config import CONFIG_FILE_NAME -from ape.plugins import plugin_manager -from ape.utils import add_padding_to_strings, github_client, load_config -from ape_plugins.utils import ModifyPluginResultHandler, PluginInstallRequest +from ape.utils import github_client, load_config +from ape_plugins.utils import ( + ModifyPluginResultHandler, + PluginMetadata, + PluginMetadataList, + PluginType, +) @click.group(short_help="Manage ape plugins") @@ -19,36 +23,20 @@ def cli(): """ -def _display_section(header: str, lines: List[Set[str]]): - click.echo(header) - for output in lines: - if output: - formatted_output = _format_output(output) - click.echo(" {}".format("\n ".join(formatted_output))) - - -def _format_output(plugins_list: Collection[str]) -> Set: - output = set() - for i in plugins_list: - text = i.replace("ape_", "") - output.add(text) - return output - - def plugins_argument(): """ An argument that is either the given list of plugins or plugins loaded from the local config file. """ - def load_from_file(ctx, file_path: Path) -> List[PluginInstallRequest]: + def load_from_file(ctx, file_path: Path) -> List[PluginMetadata]: if file_path.is_dir() and (file_path / CONFIG_FILE_NAME).is_file(): file_path = file_path / CONFIG_FILE_NAME if file_path.is_file(): config = load_config(file_path) if plugins := config.get("plugins"): - return [PluginInstallRequest.parse_obj(d) for d in plugins] + return [PluginMetadata.parse_obj(d) for d in plugins] ctx.obj.logger.warning(f"No plugins found at '{file_path}'.") return [] @@ -63,11 +51,11 @@ def callback(ctx, param, value: Tuple[str]): return ( load_from_file(ctx, file_path) if file_path.exists() - else [PluginInstallRequest(name=v) for v in value[0].split(" ")] + else [PluginMetadata(name=v) for v in value[0].split(" ")] ) else: - return [PluginInstallRequest(name=v) for v in value] + return [PluginMetadata(name=v) for v in value] return click.argument( "plugins", @@ -88,75 +76,36 @@ def upgrade_option(help: str = "", **kwargs): return click.option("-U", "--upgrade", default=False, is_flag=True, help=help, **kwargs) +def _display_all_callback(ctx, param, value): + return ( + (PluginType.CORE, PluginType.INSTALLED, PluginType.THIRD_PARTY, PluginType.AVAILABLE) + if value + else (PluginType.INSTALLED, PluginType.THIRD_PARTY) + ) + + @cli.command(name="list", short_help="Display plugins") @click.option( "-a", "--all", - "display_all", + "to_display", default=False, is_flag=True, + callback=_display_all_callback, help="Display all plugins installed and available (including Core)", ) @ape_cli_context() -def _list(cli_ctx, display_all): - installed_core_plugins = set() - installed_org_plugins = {} - installed_third_party_plugins = {} - plugin_list = plugin_manager.list_name_plugin() - spaced_names = add_padding_to_strings([p[0] for p in plugin_list], extra_spaces=4) - - for name in spaced_names: - plugin = PluginInstallRequest(name=name.strip()) - if plugin.in_core: - if not display_all: - continue +def _list(cli_ctx, to_display): + registered_plugins = cli_ctx.plugin_manager.registered_plugins + available_plugins = github_client.available_plugins + metadata = PluginMetadataList.from_package_names(registered_plugins.union(available_plugins)) + if output := metadata.to_str(include=to_display): + click.echo(output) + if not metadata.installed and not metadata.third_party: + click.echo("No plugins installed (besides core plugins).") - installed_core_plugins.add(name) - - elif plugin.is_available: - installed_org_plugins[name] = plugin.current_version - elif not plugin.in_core or not plugin.is_available: - installed_third_party_plugins[name] = plugin.current_version - else: - cli_ctx.logger.error(f"'{plugin.name}' is not a plugin.") - - sections: Dict[str, List[Set[str]]] = {} - if display_all: - sections["Installed Core Plugins"] = [installed_core_plugins] - - # Get all plugins that are available and not already installed. - available_plugins = list( - github_client.available_plugins - {p.strip() for p in installed_org_plugins.keys()} - ) - - formatted_org_plugins = {f"{k}{v}" for k, v in installed_org_plugins.items()} - formatted_installed_third_party_plugins = { - f"{k}{v}" for k, v in installed_third_party_plugins.items() - } - # Get the list of plugin lists that are populated. - installed_plugin_lists = [ - ls for ls in [formatted_org_plugins, formatted_installed_third_party_plugins] if ls - ] - if installed_plugin_lists: - sections["Installed Plugins"] = installed_plugin_lists - elif not display_all and available_plugins: - # User has no plugins installed | can't verify installed plugins - click.echo("No plugins installed. Use '--all' to see available plugins.") - - if display_all: - available_second_output = _format_output(available_plugins) - if available_second_output: - sections["Available Plugins"] = [available_second_output] - elif github_client.available_plugins: - click.echo("You have installed all the available plugins.\n") - - for i in range(len(sections)): - header = list(sections.keys())[i] - output = sections[header] - _display_section(f"{header}:", output) - - if i < len(sections) - 1: - click.echo() + else: + click.echo("No plugins installed.") @cli.command() @@ -168,44 +117,50 @@ def install(cli_ctx, plugins, skip_confirmation, upgrade): """Install plugins""" failures_occurred = False - for plugin_request in plugins: - if plugin_request.in_core: - cli_ctx.logger.error(f"Cannot install core 'ape' plugin '{plugin_request.name}'.") + for plugin in plugins: + if plugin.in_core: + cli_ctx.logger.error(f"Cannot install core 'ape' plugin '{plugin.name}'.") failures_occurred = True continue - elif plugin_request.version is not None and upgrade: + elif plugin.version is not None and upgrade: cli_ctx.logger.error( f"Cannot use '--upgrade' option when specifying " - f"a version for plugin '{plugin_request.name}'." + f"a version for plugin '{plugin.name}'." ) failures_occurred = True continue - # if plugin is installed but not a 2nd class. It must be a third party - elif not plugin_request.is_installed and not plugin_request.is_available: - cli_ctx.logger.warning(f"Plugin '{plugin_request.name}' is not an trusted plugin.") + # if plugin is installed but not trusted. It must be a third party + elif plugin.is_third_party: + cli_ctx.logger.warning(f"Plugin '{plugin.name}' is not an trusted plugin.") - result_handler = ModifyPluginResultHandler(cli_ctx.logger, plugin_request) - pip_arguments = [sys.executable, "-m", "pip", "install", "--quiet"] + result_handler = ModifyPluginResultHandler(plugin) + pip_arguments = [sys.executable, "-m", "pip", "install"] if upgrade: - cli_ctx.logger.info(f"Upgrading '{plugin_request.name}'...") - pip_arguments.extend(("--upgrade", plugin_request.package_name)) + cli_ctx.logger.info(f"Upgrading '{plugin.name}'...") + pip_arguments.extend(("--upgrade", plugin.package_name)) + + version_before = plugin.current_version + + # NOTE: There can issues when --quiet is not at the end. + pip_arguments.append("--quiet") - version_before = plugin_request.current_version result = subprocess.call(pip_arguments) # Returns ``True`` when upgraded successfully failures_occurred = not result_handler.handle_upgrade_result(result, version_before) - elif plugin_request.can_install and ( - plugin_request.is_available + elif plugin.can_install and ( + plugin.is_available or skip_confirmation - or click.confirm(f"Install unknown 3rd party plugin '{plugin_request.name}'?") + or click.confirm(f"Install the '{plugin.name}' plugin?") ): - cli_ctx.logger.info(f"Installing {plugin_request}...") - pip_arguments.append(plugin_request.install_str) + cli_ctx.logger.info(f"Installing {plugin}...") + + # NOTE: There can issues when --quiet is not at the end. + pip_arguments.extend((plugin.install_str, "--quiet")) # NOTE: Be *extremely careful* with this command, as it modifies the user's # installed packages, to potentially catastrophic results @@ -215,8 +170,7 @@ def install(cli_ctx, plugins, skip_confirmation, upgrade): else: cli_ctx.logger.warning( - f"'{plugin_request.name}' is already installed. " - f"Did you mean to include '--upgrade'." + f"'{plugin.name}' is already installed. " f"Did you mean to include '--upgrade'." ) if failures_occurred: @@ -237,7 +191,7 @@ def uninstall(cli_ctx, plugins, skip_confirmation): cli_ctx.logger.warning("Specifying a version when uninstalling is not necessary.") did_warn_about_version = True - result_handler = ModifyPluginResultHandler(cli_ctx.logger, plugin) + result_handler = ModifyPluginResultHandler(plugin) # if plugin is installed but not a 2nd class. It must be a third party if plugin.is_installed and not plugin.is_available: @@ -259,7 +213,7 @@ def uninstall(cli_ctx, plugins, skip_confirmation): skip_confirmation or click.confirm(f"Remove plugin '{plugin}'?") ): cli_ctx.logger.info(f"Uninstalling '{plugin.name}'...") - args = [sys.executable, "-m", "pip", "uninstall", "--quiet", "-y", plugin.package_name] + args = [sys.executable, "-m", "pip", "uninstall", "-y", plugin.package_name, "--quiet"] # NOTE: Be *extremely careful* with this command, as it modifies the user's # installed packages, to potentially catastrophic results diff --git a/src/ape_plugins/utils.py b/src/ape_plugins/utils.py index 8bcebb71a2..9f19d64094 100644 --- a/src/ape_plugins/utils.py +++ b/src/ape_plugins/utils.py @@ -1,42 +1,131 @@ import subprocess import sys -from typing import List, Optional, Tuple +from enum import Enum +from functools import cached_property +from typing import Iterator, List, Optional, Sequence, Set, Tuple from ape.__modules__ import __modules__ from ape._pydantic_compat import root_validator -from ape.logging import ApeLogger +from ape.logging import logger from ape.plugins import clean_plugin_name -from ape.utils import BaseInterfaceModel, cached_property, get_package_version, github_client +from ape.utils import BaseInterfaceModel, get_package_version, github_client +from ape.utils.basemodel import BaseModel # Plugins maintained OSS by ApeWorX (and trusted) CORE_PLUGINS = {p for p in __modules__ if p != "ape"} -def _pip_freeze_plugins() -> List[str]: - # NOTE: This uses 'pip' subprocess because often we have installed - # in the same process and this session's site-packages won't know about it yet. - output = subprocess.check_output([sys.executable, "-m", "pip", "freeze"]) - lines = [ - p - for p in output.decode().splitlines() - if p.startswith("ape-") or (p.startswith("-e") and "ape-" in p) - ] - - new_lines = [] - for package in lines: - if "-e" in package: - new_lines.append(package.split(".git")[0].split("/")[-1]) - elif "@" in package: - new_lines.append(package.split("@")[0].strip()) - elif "==" in package: - new_lines.append(package.split("==")[0].strip()) - else: - new_lines.append(package) +class PluginType(Enum): + CORE = "core" + """ + Plugins that ship with the core product. + """ - return new_lines + INSTALLED = "installed" + """ + Plugins that are installed (packages). + """ + THIRD_PARTY = "third-party" + """ + Plugins that are installed that are not maintained by a trusted source. + """ -class PluginInstallRequest(BaseInterfaceModel): + AVAILABLE = "available" + """ + Plugins that are available to install from a trusted-source. + """ + + +class _PipFreeze: + cache: Optional[Set[str]] = None + + def get_plugins(self, use_cache: bool = True) -> Set[str]: + if use_cache and self.cache is not None: + return self.cache + + output = subprocess.check_output([sys.executable, "-m", "pip", "freeze"]) + lines = [ + p + for p in output.decode().splitlines() + if p.startswith("ape-") or (p.startswith("-e") and "ape-" in p) + ] + + new_lines = [] + for package in lines: + if "-e" in package: + new_lines.append(package.split(".git")[0].split("/")[-1]) + elif "@" in package: + new_lines.append(package.split("@")[0].strip()) + elif "==" in package: + new_lines.append(package.split("==")[0].strip()) + else: + new_lines.append(package) + + self.cache = {x for x in new_lines} + return self.cache + + +_pip_freeze = _PipFreeze() + + +def _pip_freeze_plugins(use_cache: bool = True): + # NOTE: In a method for mocking purposes in tests. + return _pip_freeze.get_plugins(use_cache=use_cache) + + +class PluginMetadataList(BaseModel): + """ + Metadata per plugin type, including information for all plugins. + """ + + core: "PluginGroup" + available: "PluginGroup" + installed: "PluginGroup" + third_party: "PluginGroup" + + @classmethod + def from_package_names(cls, packages: Sequence[str]) -> "PluginMetadataList": + PluginMetadataList.update_forward_refs() + core = PluginGroup(plugin_type=PluginType.CORE) + available = PluginGroup(plugin_type=PluginType.AVAILABLE) + installed = PluginGroup(plugin_type=PluginType.INSTALLED) + third_party = PluginGroup(plugin_type=PluginType.THIRD_PARTY) + for name in {p for p in packages}: + plugin = PluginMetadata(name=name.strip()) + if plugin.in_core: + core.plugins.append(plugin) + elif plugin.is_available and not plugin.is_installed: + available.plugins.append(plugin) + elif plugin.is_installed and not plugin.in_core and not plugin.is_available: + third_party.plugins.append(plugin) + elif plugin.is_installed: + installed.plugins.append(plugin) + else: + logger.error(f"'{plugin.name}' is not a plugin.") + + return cls(core=core, available=available, installed=installed, third_party=third_party) + + def __str__(self) -> str: + return self.to_str() + + def to_str(self, include: Optional[Sequence[PluginType]] = None) -> str: + return str(ApePluginsRepr(self, include=include)) + + @property + def all_plugins(self) -> Iterator["PluginMetadata"]: + yield from self.core.plugins + yield from self.available.plugins + yield from self.installed.plugins + yield from self.third_party.plugins + + +def _get_available_plugins(): + # NOTE: Wrapped in a method so can GitHub HTTP can be avoided in tests. + return github_client.available_plugins + + +class PluginMetadata(BaseInterfaceModel): """ An encapsulation of a request to install a particular plugin. """ @@ -149,8 +238,11 @@ def is_installed(self) -> bool: """ ``True`` if the plugin is installed in the current Python environment. """ - ape_packages = [_split_name_and_version(n)[0] for n in _pip_freeze_plugins()] - return self.package_name in ape_packages + return self.check_installed() + + @property + def is_third_party(self) -> bool: + return self.is_installed and not self.is_available @property def pip_freeze_version(self) -> Optional[str]: @@ -179,7 +271,7 @@ def is_available(self) -> bool: Whether the plugin is maintained by the ApeWorX organization. """ - return self.module_name in github_client.available_plugins + return self.module_name in _get_available_plugins() def __str__(self): """ @@ -191,14 +283,19 @@ def __str__(self): version_key = f"=={self.version}" if self.version and self.version[0].isnumeric() else "" return f"{self.name}{version_key}" + def check_installed(self, use_cache: bool = True): + ape_packages = [ + _split_name_and_version(n)[0] for n in _pip_freeze_plugins(use_cache=use_cache) + ] + return self.package_name in ape_packages + class ModifyPluginResultHandler: - def __init__(self, logger: ApeLogger, plugin: PluginInstallRequest): - self._logger = logger + def __init__(self, plugin: PluginMetadata): self._plugin = plugin def handle_install_result(self, result) -> bool: - if not self._plugin.is_installed: + if not self._plugin.check_installed(use_cache=False): self._log_modify_failed("install") return False elif result != 0: @@ -211,7 +308,7 @@ def handle_install_result(self, result) -> bool: # Sometimes, like in editable mode, the version is missing here. plugin_id = f"{plugin_id}=={version}" - self._logger.success(f"Plugin '{plugin_id}' has been installed.") + logger.success(f"Plugin '{plugin_id}' has been installed.") return True def handle_upgrade_result(self, result, version_before: str) -> bool: @@ -222,36 +319,34 @@ def handle_upgrade_result(self, result, version_before: str) -> bool: pip_freeze_version = self._plugin.pip_freeze_version if version_before == pip_freeze_version or not pip_freeze_version: if self._plugin.version: - self._logger.info( - f"'{self._plugin.name}' already has version '{self._plugin.version}'." - ) + logger.info(f"'{self._plugin.name}' already has version '{self._plugin.version}'.") else: - self._logger.info(f"'{self._plugin.name}' already up to date.") + logger.info(f"'{self._plugin.name}' already up to date.") return True else: - self._logger.success( + logger.success( f"Plugin '{self._plugin.name}' has been " f"upgraded to version {self._plugin.pip_freeze_version}." ) return True def handle_uninstall_result(self, result) -> bool: - if self._plugin.is_installed: + if self._plugin.check_installed(use_cache=False): self._log_modify_failed("uninstall") return False elif result != 0: self._log_errors_occurred("uninstalling") return False else: - self._logger.success(f"Plugin '{self._plugin.name}' has been uninstalled.") + logger.success(f"Plugin '{self._plugin.name}' has been uninstalled.") return True def _log_errors_occurred(self, verb: str): - self._logger.error(f"Errors occurred when {verb} '{self._plugin}'.") + logger.error(f"Errors occurred when {verb} '{self._plugin}'.") def _log_modify_failed(self, verb: str): - self._logger.error(f"Failed to {verb} plugin '{self._plugin}.") + logger.error(f"Failed to {verb} plugin '{self._plugin}.") def _split_name_and_version(value: str) -> Tuple[str, Optional[str]]: @@ -264,3 +359,94 @@ def _split_name_and_version(value: str) -> Tuple[str, Optional[str]]: index = min(value.index(c) for c in chars) return value[:index], value[index:] + + +class PluginGroup(BaseModel): + """ + A group of plugin metadata by type. + """ + + plugin_type: PluginType + plugins: List[PluginMetadata] = [] + + def __bool__(self) -> bool: + return len(self.plugins) > 0 + + def __repr__(self) -> str: + return f"<{self.name} Plugins Group>" + + def __str__(self) -> str: + return self.to_str() + + @property + def name(self) -> str: + return self.plugin_type.value.capitalize() + + @property + def plugin_names(self) -> List[str]: + return [x.name for x in self.plugins] + + def to_str(self, max_length: Optional[int] = None, include_version: bool = True) -> str: + title = f"{self.name} Plugins" + if len(self.plugins) <= 0: + return title + + lines = [title] + max_length = self.max_name_length if max_length is None else max_length + plugins_sorted = sorted(self.plugins, key=lambda p: p.name) + for plugin in plugins_sorted: + line = plugin.name + if include_version: + version = plugin.version or get_package_version(plugin.package_name) + if version: + spacing = (max_length - len(line)) + 4 + line = f"{line}{spacing * ' '}{version}" + + lines.append(f" {line}") # Indent. + + return "\n".join(lines) + + @property + def max_name_length(self) -> int: + if not self.plugins: + return 0 + + return max(len(x.name) for x in self.plugins) + + +class ApePluginsRepr: + """ + A str-builder for all installed Ape plugins. + """ + + def __init__( + self, metadata: PluginMetadataList, include: Optional[Sequence[PluginType]] = None + ): + self.include = include or (PluginType.INSTALLED, PluginType.THIRD_PARTY) + self.metadata = metadata + + def __repr__(self) -> str: + to_display_str = ", ".join([x.value for x in self.include]) + return f"" + + def __str__(self) -> str: + sections = [] + + if PluginType.CORE in self.include and self.metadata.core: + sections.append(self.metadata.core) + if PluginType.INSTALLED in self.include and self.metadata.installed: + sections.append(self.metadata.installed) + if PluginType.THIRD_PARTY in self.include and self.metadata.third_party: + sections.append(self.metadata.third_party) + if PluginType.AVAILABLE in self.include and self.metadata.available: + sections.append(self.metadata.available) + + # Use a single max length for all the sections. + max_length = max(x.max_name_length for x in sections) + + version_skips = (PluginType.CORE, PluginType.AVAILABLE) + formatted_sections = [ + x.to_str(max_length=max_length, include_version=x.plugin_type not in version_skips) + for x in sections + ] + return "\n\n".join(formatted_sections) diff --git a/tests/functional/test_plugins.py b/tests/functional/test_plugins.py index a6896433b7..d18cc0dbbd 100644 --- a/tests/functional/test_plugins.py +++ b/tests/functional/test_plugins.py @@ -1,59 +1,145 @@ +from typing import Set + import pytest -from ape_plugins.utils import PluginInstallRequest +from ape_plugins.utils import ApePluginsRepr, PluginMetadata, PluginMetadataList, PluginType + +CORE_PLUGINS = ("run",) +AVAILABLE_PLUGINS = ("available", "installed") +INSTALLED_PLUGINS = ("installed", "thirdparty") +THIRD_PARTY = ("thirdparty",) +VERSION = "0.6.2" + + +@pytest.fixture(autouse=True) +def plugin_test_env(mocker): + root = "ape_plugins.utils" + + # Prevent calling out to GitHub + gh_mock = mocker.patch(f"{root}._get_available_plugins") + gh_mock.return_value = {f"ape_{x}" for x in AVAILABLE_PLUGINS} + + # Prevent requiring plugins to be installed. + installed_mock = mocker.patch(f"{root}._pip_freeze_plugins") + installed_mock.return_value = {f"ape-{x}" for x in INSTALLED_PLUGINS} + + # Prevent version lookups. + version_mock = mocker.patch(f"{root}.get_package_version") + version_mock.return_value = VERSION + + +@pytest.fixture +def package_names() -> Set[str]: + return { + f"ape-{x}" for x in [*CORE_PLUGINS, *AVAILABLE_PLUGINS, *INSTALLED_PLUGINS, *THIRD_PARTY] + } + + +@pytest.fixture +def plugin_metadata(package_names) -> PluginMetadataList: + return PluginMetadataList.from_package_names(package_names) -EXPECTED_PLUGIN_NAME = "plugin_name" +class TestPluginMetadataList: + def test_from_package_names(self, plugin_metadata): + actual = plugin_metadata + assert actual.core.plugin_names == list(CORE_PLUGINS) + assert actual.third_party.plugin_names == list(THIRD_PARTY) + assert actual.installed.plugin_names == [INSTALLED_PLUGINS[0]] # Not 3rd party + assert actual.available.plugin_names == [AVAILABLE_PLUGINS[0]] # Not installed -class TestPluginInstallRequest: + def test_all_plugins(self, plugin_metadata, package_names): + actual = {f"ape-{x.name}" for x in plugin_metadata.all_plugins} + assert actual == package_names + + +class TestPluginMetadata: @pytest.mark.parametrize( "name", ("ape-foo-bar", "ape-foo-bar", "ape_foo_bar", "foo-bar", "foo_bar") ) def test_names(self, name): - request = PluginInstallRequest(name=name) - assert request.name == "foo-bar" - assert request.package_name == "ape-foo-bar" - assert request.module_name == "ape_foo_bar" + metadata = PluginMetadata(name=name) + assert metadata.name == "foo-bar" + assert metadata.package_name == "ape-foo-bar" + assert metadata.module_name == "ape_foo_bar" - def test_parse_obj_when_version_included_with_name(self): + def test_model_when_version_included_with_name(self): # This allows parsing requirements files easier - request = PluginInstallRequest(name="ape-foo-bar==0.5.0") - assert request.name == "foo-bar" - assert request.version == "==0.5.0" + metadata = PluginMetadata(name="ape-foo-bar==0.5.0") + assert metadata.name == "foo-bar" + assert metadata.version == "==0.5.0" @pytest.mark.parametrize("version", ("0.5.0", "v0.5.0", "0.5.0a123")) def test_version(self, version): - request = PluginInstallRequest(name="foo", version=version) - assert request.version == version + metadata = PluginMetadata(name="foo", version=version) + assert metadata.version == version def test_install_str_without_version(self): - request = PluginInstallRequest(name="foo-bar") - actual = request.install_str + metadata = PluginMetadata(name="foo-bar") + actual = metadata.install_str assert actual == "ape-foo-bar" def test_install_str_with_version(self): - request = PluginInstallRequest(name="foo-bar", version="0.5.0") - actual = request.install_str + metadata = PluginMetadata(name="foo-bar", version="0.5.0") + actual = metadata.install_str assert actual == "ape-foo-bar==0.5.0" def test_install_str_with_complex_constraint(self): - request = PluginInstallRequest(name="foo", version=">=0.5.0,<0.6.0") - actual = request.install_str + metadata = PluginMetadata(name="foo", version=">=0.5.0,<0.6.0") + actual = metadata.install_str assert actual == "ape-foo>=0.5.0,<0.6.0" def test_install_str_with_complex_constraint_in_name(self): - request = PluginInstallRequest(name="foo>=0.5.0,<0.6.0") - actual = request.install_str + metadata = PluginMetadata(name="foo>=0.5.0,<0.6.0") + actual = metadata.install_str assert actual == "ape-foo>=0.5.0,<0.6.0" def test_install_str_when_using_git_remote(self): url = "git+https://example.com/ape-foo/branch" - request = PluginInstallRequest(name="foo", version=url) - actual = request.install_str + metadata = PluginMetadata(name="foo", version=url) + actual = metadata.install_str assert actual == url def test_install_str_remote_in_name(self): url = "git+https://example.com/ape-foo/branch" - request = PluginInstallRequest(name=f"foo@{url}") - actual = request.install_str + metadata = PluginMetadata(name=f"foo@{url}") + actual = metadata.install_str assert actual == url + + def test_is_available(self): + metadata = PluginMetadata(name=list(AVAILABLE_PLUGINS)[0]) + assert metadata.is_available + metadata = PluginMetadata(name="foobar") + assert not metadata.is_available + + +class TestPluginDisplaySectionMap: + def test_str(self, plugin_metadata): + plugin_map = ApePluginsRepr(plugin_metadata) + actual = str(plugin_map) + expected = f""" +Installed Plugins + installed {VERSION} + +Third-party Plugins + thirdparty {VERSION} + """ + assert actual == expected.strip() + + def test_str_all_types(self, plugin_metadata): + plugin_map = ApePluginsRepr(plugin_metadata, include=list(PluginType)) + actual = str(plugin_map) + expected = f""" +Core Plugins + run + +Installed Plugins + installed {VERSION} + +Third-party Plugins + thirdparty {VERSION} + +Available Plugins + available + """ + assert actual == expected.strip() diff --git a/tests/integration/cli/conftest.py b/tests/integration/cli/conftest.py index 446ae18ddb..bf6d8bf7cc 100644 --- a/tests/integration/cli/conftest.py +++ b/tests/integration/cli/conftest.py @@ -228,7 +228,7 @@ def __init__(self): def invoke_list(self, arguments: Optional[List] = None): arguments = arguments or [] result = self.invoke(["list", *arguments]) - assert result.exit_code == 0 + assert result.exit_code == 0, result.output return ListResult.parse_output(result.output) return PluginSubprocessRunner() diff --git a/tests/integration/cli/test_plugins.py b/tests/integration/cli/test_plugins.py index e8599dc8db..c3f273ee61 100644 --- a/tests/integration/cli/test_plugins.py +++ b/tests/integration/cli/test_plugins.py @@ -17,9 +17,9 @@ def __init__(self, header: str, lines: List[str]): class ListResult: - CORE_KEY = "Installed Core Plugins:" - INSTALLED_KEY = "Installed Plugins:" - AVAILABLE_KEY = "Available Plugins:" + CORE_KEY = "Core Plugins" + INSTALLED_KEY = "Installed Plugins" + AVAILABLE_KEY = "Available Plugins" def __init__(self, lines: List[str]): self._lines = lines