diff --git a/docs/tutorials/plugins.md b/docs/tutorials/plugins.md index 19fc4300d..71c956c60 100644 --- a/docs/tutorials/plugins.md +++ b/docs/tutorials/plugins.md @@ -9,6 +9,10 @@ To enable a plugin, run `odev plugin --enable /`. - [Plugins](#plugins) - [Table of contents](#table-of-contents) + - [Finding and managing plugins](#finding-and-managing-plugins) + - [Searching for plugins](#searching-for-plugins) + - [Listing local plugins](#listing-local-plugins) + - [Inspecting a plugin](#inspecting-a-plugin) - [Creating a new plugin](#creating-a-new-plugin) - [Plugin structure](#plugin-structure) - [The manifest](#the-manifest) @@ -18,6 +22,64 @@ To enable a plugin, run `odev plugin --enable /`. - [Adding a new command](#adding-a-new-command) - [Extending a command](#extending-a-command) +## Finding and managing plugins + +### Searching for plugins + +Run `odev plugin --search` to look for plugins published on GitHub. Odev queries the GitHub search API for the `odev` +and `plugin` keywords, then keeps only the repositories exposing a valid [manifest](#the-manifest) at their root, so +unrelated repositories never show up in the results. + +Add a term to narrow the search down, quoting it if it contains several words: + +```sh +odev plugin --search editor +odev plugin --search "upgrade code" +``` + +The results are displayed in a table showing, for each plugin, the version declared on its default branch, its number +of stars and whether it is already available locally. Archived repositories and the +[template repository](https://github.com/odoo-odev/odev-plugin-template), which cannot be installed, are left out. +`--limit` caps how many repositories are inspected (20 by default) to stay within the GitHub API rate limits. + +> [!NOTE] +> +> Searching never installs anything. Copy the name of a plugin from the results and run +> `odev plugin --enable /` to install it. + +### Listing local plugins + +Run `odev plugin --list` to display every plugin available on your machine, in one of the following states: + +| State | Meaning | +| ---------- | --------------------------------------------------------------------------------------------- | +| `enabled` | The plugin is loaded by odev. | +| `shadowed` | The plugin is enabled but another plugin already uses its module name, so it cannot be loaded. | +| `missing` | The plugin is enabled but its link under `~/.config/odev/plugins` is gone. | +| `disabled` | The plugin was downloaded previously but is not enabled; re-enabling it will not clone it again. | + +### Inspecting a plugin + +Use `odev plugin --show /` to get the details of a single plugin: its state, version, +branch, path, dependencies and description. Without an argument, `--show` details every plugin available locally, in +the same order as `--list`. + +A plugin that is not on your machine is looked up on GitHub, so `--show` also describes plugins you have not installed +yet, or that you uninstalled and whose clone you deleted: + +```sh +odev plugin --show odoo-odev/odev-plugin-editor-vscode +``` + +Plugins already available locally are read from disk and never trigger a request to GitHub. The full +`/` name is required to look a plugin up remotely: a repository name on its own is only +matched against the plugins present on your machine, as long as it is not ambiguous. + +> [!NOTE] +> +> When GitHub cannot be reached — no token configured, no network, rate limit exceeded — `--show` silently falls back +> to the information available locally instead of failing. + ## Creating a new plugin To create a new odev plugin, start by creating a new repository. You can copy a @@ -52,6 +114,9 @@ Create a new file `__manifest__.py` at the root of your plugin with the followin repository). Replace the docstring by a summary of your module's features. This will be read by Odev and displayed when required by the `odev plugin` command. +The `__version__` assignment is also what makes a repository recognizable as a plugin: a repository without a root +`__manifest__.py` declaring it is ignored by `odev plugin --search`. + If any, add the dependencies (other plugins) of your own plugin. For example, `odoo-odev/odev-plugin-editor-vscode` depends on the abstract plugin `odoo-odev/odev-plugin-editor-base` which is therefore required for the plugin to work: [VScode Editor plugin's depends](https://github.com/odoo-odev/odev-plugin-editor-vscode/blob/main/__manifest__.py#L38). diff --git a/odev/_version.py b/odev/_version.py index 06d8cbf0d..6a030787c 100644 --- a/odev/_version.py +++ b/odev/_version.py @@ -22,4 +22,4 @@ # or merged change. # ------------------------------------------------------------------------------ -__version__ = "4.30.4" +__version__ = "4.31.0" diff --git a/odev/commands/utilities/plugin.py b/odev/commands/utilities/plugin.py index ec506f4c3..6038e2add 100644 --- a/odev/commands/utilities/plugin.py +++ b/odev/commands/utilities/plugin.py @@ -1,23 +1,127 @@ -"""Enable and disable plugins to add new features and commands.""" +"""Search, enable and disable plugins to add new features and commands.""" -from typing import cast +import textwrap +from collections.abc import Mapping +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Any -from odev.common import args, string +from git import InvalidGitRepositoryError, NoSuchPathError +from github import GithubException, RateLimitExceededException +from github.Repository import Repository + +from odev.common import args, progress, string from odev.common.commands import Command -from odev.common.connectors import GitConnector +from odev.common.connectors import GitConnector, GithubConnector +from odev.common.connectors.git import GITHUB_DOMAIN, GITHUB_SEARCH_DEFAULT_LIMIT +from odev.common.console import TableHeader +from odev.common.errors import ConnectorError from odev.common.logging import logging -from odev.common.odev import Plugin +from odev.common.odev import ( + PLUGIN_MANIFEST_FILENAME, + Manifest, + parse_plugin_manifest, + plugin_module_name, +) logger = logging.getLogger(__name__) +SEARCH_KEYWORDS = "odev plugin" +"""Keywords a repository must match to be considered a candidate odev plugin. +The `odev` keyword alone is far too noisy to be usable, as it collides with unrelated repositories. +""" + +SEARCH_QUALIFIERS = "in:name,description,topics" +"""Fields the search keywords are matched against on GitHub.""" + +EXCLUDED_REPOSITORIES = frozenset({"odoo-odev/odev-plugin-template"}) +"""Repositories exposing a plugin manifest but that cannot be installed, such as the template repository +used as a starting point to create new plugins. +""" + +STATE_ENABLED = "enabled" +"""The plugin is linked under the plugins directory and loaded by the framework.""" + +STATE_SHADOWED = "shadowed" +"""The plugin is enabled in the configuration but another plugin already uses its module name.""" + +STATE_MISSING = "missing" +"""The plugin is enabled in the configuration but is not linked under the plugins directory.""" + +STATE_DISABLED = "disabled" +"""The plugin is available locally but is not enabled.""" + +STATE_NOT_DOWNLOADED = "not downloaded" +"""The plugin exists on GitHub but is not available locally.""" + +STATE_STYLES: Mapping[str, str] = { + STATE_ENABLED: "color.green", + STATE_SHADOWED: "color.yellow", + STATE_MISSING: "color.red", + STATE_DISABLED: "color.black", + STATE_NOT_DOWNLOADED: "color.black", +} +"""Style used to render each possible state of a plugin.""" + +STATE_ORDER: tuple[str, ...] = ( + STATE_ENABLED, + STATE_SHADOWED, + STATE_MISSING, + STATE_DISABLED, + STATE_NOT_DOWNLOADED, +) +"""Order in which plugins are sorted in tables, most relevant states first.""" + +DESCRIPTION_MAX_WIDTH = 60 +"""Maximum width of a plugin description before it gets truncated in tables.""" + +NAME_MIN_WIDTH = 32 +"""Minimum width reserved for plugin names in tables, as they are needed to enable a plugin.""" + + +@dataclass(frozen=True) +class PluginInfo: + """Everything known about a plugin, whether it is available locally or only on GitHub.""" + + name: str + """Name of the plugin, in the format `organization/repository`.""" + + state: str + """Current state of the plugin, one of the `STATE_*` constants.""" + + version: str = "" + """Version number declared in the manifest of the plugin.""" + + branch: str = "" + """Branch currently checked out in the local clone of the plugin.""" + + path: Path | None = None + """Path to the local clone of the plugin, if it exists.""" + + depends: list[str] = field(default_factory=list) + """Other plugins this one depends on.""" + + description: str = "" + """Description of the plugin, taken from the docstring of its manifest.""" + + shadowed_by: str = "" + """Plugin already using the module name of this one, preventing it from being loaded.""" + + stars: int | None = None + """Number of stars of the repository, `None` if it was not looked up on GitHub.""" + + archived: bool = False + """Whether the repository is archived and no longer maintained.""" + + class PluginCommand(Command): - """Enable and disable plugins to add new features and commands.""" + """Search, enable and disable plugins to add new features and commands.""" _name = "plugin" _aliases = ["plugins"] - _exclusive_arguments = [("enable", "disable", "show")] + _exclusive_arguments = [("enable", "disable", "show", "search", "list")] enable = args.Flag(aliases=["-e", "--enable"], description="Download and enable an inactive plugin.") disable = args.Flag(aliases=["-d", "--disable"], description="Disable an active plugin.") @@ -25,10 +129,27 @@ class PluginCommand(Command): aliases=["-s", "--show"], description="Show the state of a plugin and its description if available.", ) + search = args.Flag( + aliases=["-S", "--search"], + description="""Search GitHub for plugins matching the given terms. + Only repositories exposing a valid plugin manifest are listed. + """, + ) + action_list = args.Flag( + name="list", + aliases=["-l", "--list"], + description="List the plugins available locally, whether they are enabled or not.", + ) + limit = args.Integer( + aliases=["-n", "--limit"], + default=GITHUB_SEARCH_DEFAULT_LIMIT, + description="Maximum number of repositories to inspect when searching for plugins.", + ) plugin = args.String( description="""Plugin to enable or disable, must be a git repository hosted on GitHub. Use format /. - If `--show` is used and no plugin is provided, show the state of all plugins. + If `--show` is used and no plugin is provided, show the state of all enabled plugins. + If `--search` is used, this is the term to search for; quote it to search for multiple terms. """, nargs="?", ) @@ -36,58 +157,465 @@ class PluginCommand(Command): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - if not self.args.plugin and not self.args.show: + self.__local_plugins: list[PluginInfo] | None = None + """Cached list of the plugins known locally.""" + + if not self.args.plugin and not (self.args.show or self.args.search or self.args.list): raise self.error("Missing argument: plugin") def run(self): - """Enable or disable a plugin.""" - if self.args.show: - if self.args.plugin: - self.__show_plugin_info(self.args.plugin.split("/")[-1]) - return + """Search, list, enable or disable plugins.""" + if self.args.search: + self.search_plugins() - for plugin in self.odev.plugins: - self.__show_plugin_info(plugin.name) - self.console.print() + if self.args.list: + self.list_plugins() + + if self.args.show: + self.show_plugins() if self.args.enable: - self.odev.install_plugin(self.args.plugin) - return + self.odev.install_plugin(self.__resolve_plugin_name(self.args.plugin)) if self.args.disable: - self.odev.uninstall_plugin(self.args.plugin) + self.odev.uninstall_plugin(self.__resolve_plugin_name(self.args.plugin)) + + # --- Searching plugins on GitHub ------------------------------------------ + + def search_plugins(self) -> None: + """Search GitHub for odev plugins and display the results in a table.""" + terms = self.args.plugin or "" + query = " ".join(part for part in (SEARCH_KEYWORDS, terms, SEARCH_QUALIFIERS) if part) + states = {plugin.name: plugin.state for plugin in self._discover_plugins()} + github = GithubConnector() + rows: list[list[Any]] = [] + + with progress.spinner(f"Searching GitHub for plugins matching {SEARCH_KEYWORDS} {terms}".strip()) as status: + for repository in self.__search_repositories(github, query): + status.update(f"Inspecting repository {repository.full_name!r}") + manifest = self.__remote_manifest(github, repository) + + if manifest is None: + continue + + state = states.get(repository.full_name, STATE_NOT_DOWNLOADED) + rows.append( + [ + string.link(repository.full_name, repository.html_url), + manifest["version"], + str(repository.stargazers_count), + string.stylize(state, STATE_STYLES[state]), + self.__shorten(manifest["description"] or repository.description or ""), + ] + ) + + if not rows: + raise self.error(f"No odev plugin found matching {terms!r}" if terms else "No odev plugin found on GitHub") + + headers = [ + TableHeader("Plugin", min_width=NAME_MIN_WIDTH, style="color.purple"), + TableHeader("Version", align="right", style="repr.version"), + TableHeader("Stars", align="right"), + TableHeader("State"), + TableHeader("Description"), + ] + + self.print() + self.table(headers, rows, title=f"Plugins matching {terms!r}" if terms else "Plugins") + self.console.clear_line() + + logger.info("Run 'odev plugin --enable /' to install one of these plugins") + + def __search_repositories(self, github: GithubConnector, query: str) -> list[Repository]: + """Query the GitHub search API and discard the repositories that cannot be plugins. + + :param github: Connector to the GitHub API. + :param query: The search query, using the GitHub search syntax. + :return: The candidate repositories to inspect. + """ + try: + repositories = github.search_repositories(query, limit=self.args.limit) + except RateLimitExceededException as error: + raise self.error("GitHub API rate limit exceeded, please try again in a few minutes") from error + except GithubException as error: + raise self.error(f"Failed to search for plugins on GitHub: {error}") from error + + return [ + repository + for repository in repositories + if not repository.archived and repository.full_name not in EXCLUDED_REPOSITORIES + ] + + def __remote_manifest(self, github: GithubConnector, repository: Repository) -> Manifest | None: + """Fetch and parse the manifest of a remote repository to check whether it is an odev plugin. + + :param github: Connector to the GitHub API. + :param repository: The remote repository to inspect. + :return: The manifest of the plugin, or `None` if the repository is not an odev plugin. + """ + source = github.get_repository_file(repository, PLUGIN_MANIFEST_FILENAME) + + if source is None: + return None + + return parse_plugin_manifest(source, repository.full_name) + + # --- Listing local plugins ------------------------------------------------ + + def list_plugins(self) -> None: + """List the plugins known locally and display them in a table.""" + plugins = self._discover_plugins() + + if not plugins: + raise self.error("No plugin found locally") + + headers = [ + TableHeader("Plugin", min_width=NAME_MIN_WIDTH, style="color.purple"), + TableHeader("Version", align="right", style="repr.version"), + TableHeader("Branch", style="color.cyan"), + TableHeader("State"), + TableHeader("Depends"), + ] + rows = [ + [ + string.link(plugin.name, f"https://{GITHUB_DOMAIN}/{plugin.name}"), + plugin.version, + plugin.branch, + string.stylize(plugin.state, STATE_STYLES[plugin.state]), + ", ".join(dependency.split("/")[-1] for dependency in plugin.depends), + ] + for plugin in plugins + ] + + self.print() + self.table(headers, rows, title="Plugins") + self.console.clear_line() + + if shadowed := [plugin for plugin in plugins if plugin.state == STATE_SHADOWED]: + logger.warning( + "The following plugins are enabled but cannot be loaded as another plugin already uses their " + "module name:\n" + + string.join_bullet([f"{plugin.name} (shadowed by {plugin.shadowed_by})" for plugin in shadowed]) + ) + + # --- Showing a single plugin ---------------------------------------------- + + def show_plugins(self) -> None: + """Show detailed information about one plugin, or about all the plugins available locally.""" + if self.args.plugin: + self.__show_plugin_info(self.__resolve_plugin_name(self.args.plugin)) return - def __show_plugin_info(self, plugin_name: str): - """Show the plugin information. + for plugin in self._discover_plugins(): + self.__show_plugin_info(plugin.name) + self.console.print() - :param plugin_name: The name of the plugin to show information for. + def __show_plugin_info(self, name: str) -> None: + """Show the state of a plugin and its description if available. + + :param name: The name of the plugin, in the format `organization/repository`. """ - if plugin_name not in self.config.plugins.enabled: - logger.info(f"Plugin {plugin_name!r} is {string.stylize('disabled', 'color.red')}") - else: - plugin = self.__get_plugin(plugin_name) - plugin_git = GitConnector(plugin_name) + plugin = self.__plugin_info(name) + + if plugin is None: logger.info( - string.normalize_indent( - f""" - Plugin {plugin.name!r} is {string.stylize("enabled", "color.green")} - {string.stylize("Version:", "color.black")} {string.stylize(plugin.manifest["version"], "repr.version")} - {string.stylize("Branch:", "color.black")} {string.stylize(cast(str, plugin_git.branch), "color.cyan")} - {string.stylize("Path:", "color.black")} {plugin.path.resolve()} - """ - ) + f"Plugin {name!r} is {string.stylize(STATE_NOT_DOWNLOADED, STATE_STYLES[STATE_NOT_DOWNLOADED])}" + ) + + if "/" not in name: + logger.info("Use the full name of the plugin as '/' to look it up on GitHub") + + return + + fields = [ + ("Version", string.stylize(plugin.version, "repr.version") if plugin.version else ""), + ("Branch", string.stylize(plugin.branch, "color.cyan") if plugin.branch else ""), + ("Stars", "" if plugin.stars is None else str(plugin.stars)), + ("Path", plugin.path.as_posix() if plugin.path is not None else ""), + ("Depends", ", ".join(plugin.depends)), + ("URL", f"https://{GITHUB_DOMAIN}/{plugin.name}"), + ] + width = max(len(label) for label, value in fields if value) + 1 + details = [f"Plugin {plugin.name!r} is {string.stylize(plugin.state, STATE_STYLES[plugin.state])}"] + details.extend( + f"{string.stylize(f'{label}:'.ljust(width), 'color.black')} {value}" for label, value in fields if value + ) + logger.info("\n".join(details)) + + if plugin.description: + self.console.print() + self.console.print(string.indent(plugin.description, 4).rstrip("\n")) + + self.__warn_plugin_unusable(plugin) + + def __warn_plugin_unusable(self, plugin: PluginInfo) -> None: + """Warn about the reasons a plugin cannot be loaded or installed, and tell how to install it otherwise. + + :param plugin: The plugin the information of which is being displayed. + """ + if plugin.state == STATE_SHADOWED: + logger.warning(f"Plugin {plugin.name!r} is shadowed by {plugin.shadowed_by!r} and cannot be loaded") + + if plugin.archived: + logger.warning(f"Repository {plugin.name!r} is archived and is not maintained anymore") + + if plugin.name in EXCLUDED_REPOSITORIES: + logger.warning( + f"Repository {plugin.name!r} is a template used to create new plugins and cannot be installed" ) + elif plugin.state == STATE_NOT_DOWNLOADED: + logger.info(f"Run 'odev plugin --enable {plugin.name}' to install this plugin") + + def __plugin_info(self, name: str) -> PluginInfo | None: + """Gather everything known about a plugin, completing local information with GitHub when needed. + + Plugins available locally are never looked up on GitHub, so this stays offline for the common case. + + :param name: The name of the plugin, in the format `organization/repository`. + :return: The plugin, or `None` if nothing is known about it. + """ + plugin = next((known for known in self._discover_plugins() if known.name == name), None) + + if (plugin is not None and plugin.version) or "/" not in name: + return plugin - if plugin.manifest["description"]: - self.console.print() - self.console.print(string.indent(cast(str, plugin.manifest["description"]), 4).rstrip("\n")) + remote = self.__fetch_remote_plugin(name) - def __get_plugin(self, name: str) -> Plugin: - """Find a plugin by its name.""" - plugin = next((plugin for plugin in self.odev.plugins if plugin[0] == name), None) + if remote is None: + return plugin if plugin is None: - raise self.error(f"Plugin {name!r} not found or could not be loaded") + return remote + + return replace( + plugin, + version=remote.version, + branch=plugin.branch or remote.branch, + depends=remote.depends, + description=remote.description, + stars=remote.stars, + archived=remote.archived, + ) + + def __fetch_remote_plugin(self, name: str) -> PluginInfo | None: + """Fetch the details of a plugin from GitHub, without cloning it. + + Failing to reach GitHub is not an error: the plugin is then only described by what is known locally. + + :param name: The name of the plugin, in the format `organization/repository`. + :return: The plugin as published on GitHub, or `None` if it could not be fetched. + """ + try: + with progress.spinner(f"Fetching plugin {name!r} from GitHub"): + github = GithubConnector() + repository = github.get_repository(name) + + if repository is None: + return None + + manifest = self.__remote_manifest(github, repository) + except (ConnectorError, GithubException) as error: + logger.debug(f"Could not fetch plugin {name!r} from GitHub: {error}") + return None + + if manifest is None: + logger.debug(f"Repository {name!r} does not expose a valid plugin manifest") + return None + + return PluginInfo( + name=repository.full_name, + state=STATE_NOT_DOWNLOADED, + version=manifest["version"], + branch=repository.default_branch, + depends=list(manifest["depends"]), + description=manifest["description"] or repository.description or "", + stars=repository.stargazers_count, + archived=repository.archived, + ) + + # --- Discovering local plugins -------------------------------------------- + + def _discover_plugins(self) -> list[PluginInfo]: + """List all the plugins known locally, whether they are enabled, disabled or broken. + + :return: The plugins known locally, sorted by state then by name. + """ + if self.__local_plugins is None: + plugins = self.__discover_enabled() + plugins.update(self.__discover_configured(plugins)) + plugins.update(self.__discover_downloaded(plugins)) + + self.__local_plugins = sorted( + plugins.values(), + key=lambda plugin: (STATE_ORDER.index(plugin.state), plugin.name), + ) + + return self.__local_plugins + + def __discover_enabled(self) -> dict[str, PluginInfo]: + """List the plugins linked under the plugins directory and loaded by the framework. + + :return: The enabled plugins, mapped by name. + """ + plugins: dict[str, PluginInfo] = {} + + for plugin in self.odev.plugins: + resolved_path = plugin.path.resolve() + + if plugin.path.name.startswith((".", "_")) or not resolved_path.is_dir(): + continue + + plugins[plugin.name] = PluginInfo( + name=plugin.name, + state=STATE_ENABLED, + version=plugin.manifest["version"], + branch=self.__repository_branch(resolved_path), + path=resolved_path, + depends=list(plugin.manifest["depends"]), + description=(plugin.manifest["description"] or "").strip(), + ) + + return plugins + + def __discover_configured(self, enabled: Mapping[str, PluginInfo]) -> dict[str, PluginInfo]: + """List the plugins enabled in the configuration that the framework could not load. + + :param enabled: The plugins already discovered as enabled, mapped by name. + :return: The plugins whose link is either missing or taken by another plugin, mapped by name. + """ + modules = {plugin_module_name(name): name for name in enabled} + plugins: dict[str, PluginInfo] = {} - return plugin + for name in self.config.plugins.enabled: + if name in enabled: + continue + + shadowed_by = modules.get(plugin_module_name(name), "") + plugins[name] = self.__build_plugin( + name, + self.config.paths.repositories / name, + STATE_SHADOWED if shadowed_by else STATE_MISSING, + shadowed_by=shadowed_by, + ) + + return plugins + + def __discover_downloaded(self, known: Mapping[str, PluginInfo]) -> dict[str, PluginInfo]: + """List the plugins cloned locally but neither enabled nor referenced in the configuration. + + Repositories that cannot be installed are left out so they are never advertised as available, but a plugin + already enabled is always reported, whatever its repository. + + :param known: The plugins already discovered, mapped by name. + :return: The plugins available locally but not enabled, mapped by name. + """ + repositories_path = self.config.paths.repositories + plugins: dict[str, PluginInfo] = {} + + if not repositories_path.is_dir(): + return plugins + + for manifest_path in sorted(repositories_path.glob(f"*/*/{PLUGIN_MANIFEST_FILENAME}")): + path = manifest_path.parent + name = f"{path.parent.name}/{path.name}" + + if name in known or name in plugins or name in EXCLUDED_REPOSITORIES: + continue + + manifest = self.__read_manifest(path, name) + + if manifest is None: + continue + + plugins[name] = self.__build_plugin(name, path, STATE_DISABLED, manifest=manifest) + + return plugins + + def __build_plugin( + self, + name: str, + path: Path, + state: str, + manifest: Manifest | None = None, + shadowed_by: str = "", + ) -> PluginInfo: + """Build the representation of a plugin that is not loaded by the framework. + + :param name: The name of the plugin, in the format `organization/repository`. + :param path: The expected path to the local clone of the plugin. + :param state: The state of the plugin, one of the `STATE_*` constants. + :param manifest: The already parsed manifest of the plugin, read from `path` if omitted. + :param shadowed_by: The plugin already using the module name of this one, if any. + :return: The plugin as known locally. + """ + manifest = manifest or self.__read_manifest(path, name) + + return PluginInfo( + name=name, + state=state, + version=manifest["version"] if manifest else "", + branch=self.__repository_branch(path), + path=path if path.is_dir() else None, + depends=list(manifest["depends"]) if manifest else [], + description=manifest["description"] if manifest else "", + shadowed_by=shadowed_by, + ) + + def __read_manifest(self, path: Path, name: str) -> Manifest | None: + """Read the manifest of a plugin located at the given path, without executing it. + + :param path: The path to the local clone of the plugin. + :param name: The name of the plugin, in the format `organization/repository`. + :return: The manifest of the plugin, or `None` if it is not a valid odev plugin. + """ + manifest_path = path / PLUGIN_MANIFEST_FILENAME + + if not manifest_path.is_file(): + return None + + try: + source = manifest_path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + logger.debug(f"Could not read the manifest of plugin {name!r} at {manifest_path.as_posix()}") + return None + + return parse_plugin_manifest(source, name) + + def __repository_branch(self, path: Path) -> str: + """Return the branch currently checked out in the local clone of a plugin. + + :param path: The path to the local clone of the plugin. + :return: The name of the active branch, empty if the path is not a git repository. + """ + if not (path / ".git").exists(): + return "" + + try: + return GitConnector(path.as_posix()).branch or "" + except (ConnectorError, InvalidGitRepositoryError, NoSuchPathError, ValueError): + logger.debug(f"Could not determine the branch of the repository at {path.as_posix()}") + return "" + + def __resolve_plugin_name(self, name: str) -> str: + """Resolve a plugin name given on the command line to its full `organization/repository` name. + + :param name: The name of the plugin, either fully qualified or the name of its repository only. + :return: The fully qualified name of the plugin, unchanged if it could not be resolved. + """ + if "/" in name: + return name + + candidates = [plugin.name for plugin in self._discover_plugins() if plugin.name.split("/")[-1] == name] + + if len(candidates) > 1: + raise self.error(f"Plugin name {name!r} is ambiguous, use one of:\n{string.join_bullet(candidates)}") + + return candidates[0] if candidates else name + + def __shorten(self, description: str) -> str: + """Collapse a description to a single line fitting the width of a table column. + + :param description: The description to shorten. + :return: The shortened description. + """ + return textwrap.shorten(description, DESCRIPTION_MAX_WIDTH, placeholder="...") diff --git a/odev/common/connectors/__init__.py b/odev/common/connectors/__init__.py index 3523703cd..bd0f6cef2 100644 --- a/odev/common/connectors/__init__.py +++ b/odev/common/connectors/__init__.py @@ -9,16 +9,16 @@ # importing this package does not import every connector it exposes. if TYPE_CHECKING: from odev.common.connectors.base import Connector - from odev.common.connectors.git import GitConnector, GitWorktree, Stash + from odev.common.connectors.git import GitConnector, GithubConnector, GitWorktree, Stash from odev.common.connectors.postgres import PostgresConnector from odev.common.connectors.rest import RestConnector from odev.common.connectors.rpc import RpcConnector - __all__ = [ "Connector", "GitConnector", "GitWorktree", + "GithubConnector", "PostgresConnector", "RestConnector", "RpcConnector", @@ -30,6 +30,7 @@ { "Connector": "base", "GitConnector": "git", + "GithubConnector": "git", "GitWorktree": "git", "Stash": "git", "PostgresConnector": "postgres", diff --git a/odev/common/connectors/git.py b/odev/common/connectors/git.py index 47b22fa10..f570ad6e1 100644 --- a/odev/common/connectors/git.py +++ b/odev/common/connectors/git.py @@ -3,6 +3,7 @@ import re import shutil from collections.abc import Callable, Generator, Mapping, Sequence +from itertools import islice from pathlib import Path from types import FrameType from typing import ( @@ -25,6 +26,7 @@ if TYPE_CHECKING: from github import Github + from github.Repository import Repository GITHUB_DOMAIN = "github.com" @@ -36,6 +38,9 @@ GIT_EXPECTED_REPO_PARTS = 2 """The expected number of parts in a git repository name (organization/repository).""" +GITHUB_SEARCH_DEFAULT_LIMIT = 20 +"""The default maximum number of repositories returned by a search on the GitHub API.""" + # Git progress opcodes, used for progress reporting GIT_OPCODE_DOWNLOAD = 33 GIT_OPCODE_RESOLVE = 65 @@ -234,8 +239,8 @@ def pending_changes(self) -> tuple[int, int]: return commits_behind, commits_ahead -class GitConnector(Connector): - """A class for connecting to the Github API.""" +class GithubConnector(Connector): + """A class for connecting to the Github API, without any repository context.""" _token: str | None = None """The Github API token for the current session.""" @@ -243,6 +248,155 @@ class GitConnector(Connector): _connection: "Github | None" = None """The connection to the Github API.""" + @property + def url(self) -> str: + """The URL to the Github API.""" + return f"https://api.{GITHUB_DOMAIN}" + + @property + def authenticated(self) -> bool: + """Whether the current session is authenticated.""" + if self._connection is None: + return False + + from github import GithubException # noqa: PLC0415 - importing the GitHub API client is expensive + + try: + self._connection.get_user().login # noqa: B018 - login is a property + except GithubException: + return False + else: + return True + + def connect(self): + """Connect to the Github API.""" + from github import Auth as GithubAuth, Github # noqa: PLC0415 - importing the GitHub API client is expensive + + if self._token is None: + + def get_token(prompt: bool) -> str | None: + return self.store.secrets.get( + GITHUB_DOMAIN, + scope="api", + fields=["password"], + prompt_format="GitHub API token:", + ask_missing=prompt, + ).password + + token = get_token(prompt=False) + + if not token: + logger.info( + """Connection to your GitHub account is necessary to pursue this operation, please configure a + Personal Access Token (classic) on https://github.com/settings/tokens with the following permissions: + - repo (all) + - user: + - read:user + - user:email + """ + ) + token = get_token(prompt=True) + + self._token = token + + if not self.connected: + self._connection = Github(auth=GithubAuth.Token(self._token)) # type: ignore [assignment] + + if not self.authenticated: + logger.warning("Failed to connect to Github API, please check your token is valid") + self.store.secrets.invalidate(GITHUB_DOMAIN, scope="api") + self._disconnect() + self.connect() + return + + logger.debug("Connected to Github API") + + def disconnect(self): + """Disconnect from the Github API.""" + self._disconnect() + logger.debug("Disconnected from Github API") + + def _disconnect(self): + """Disconnect from the Github API.""" + self._token = None + del self._connection + + def search_repositories( + self, + query: str, + limit: int = GITHUB_SEARCH_DEFAULT_LIMIT, + sort: str | None = None, + order: str = "desc", + ) -> "list[Repository]": + """Search for repositories on GitHub. + + :param query: The search query, using the GitHub search syntax. + :param limit: The maximum number of repositories to return. + :param sort: How to sort the results, one of `stars`, `forks` or `updated`; defaults to best match. + :param order: The direction of the sort, one of `asc` or `desc`. + :return: The repositories matching the search query. + """ + if limit <= 0: + return [] + + with self: + results = cast("Github", self._connection).search_repositories( + query, + **({"sort": sort, "order": order} if sort else {}), + ) + + return list(islice(results, limit)) + + def get_repository(self, name: str) -> "Repository | None": + """Fetch a repository from GitHub by its full name, without cloning it. + + :param name: The full name of the repository, in the format `organization/repository`. + :return: The repository, or `None` if it does not exist or cannot be accessed. + """ + from github import GithubException, UnknownObjectException # noqa: PLC0415 - the GitHub client is expensive + + with self: + try: + return cast("Github", self._connection).get_repo(name) + except UnknownObjectException: + return None + except GithubException as error: + logger.debug(f"Failed to fetch repository {name!r}: {error}") + return None + + @staticmethod + def get_repository_file(repository: "Repository", path: str, ref: str | None = None) -> str | None: + """Fetch the content of a file inside a remote repository, without cloning it. + + :param repository: The remote repository to fetch the file from. + :param path: The path to the file, relative to the root of the repository. + :param ref: The branch, tag or commit to fetch the file from; defaults to the default branch. + :return: The decoded content of the file, or `None` if it does not exist or cannot be read. + """ + from github import GithubException, UnknownObjectException # noqa: PLC0415 - the GitHub client is expensive + + try: + contents = repository.get_contents(path, **({"ref": ref} if ref else {})) + except UnknownObjectException: + return None + except GithubException as error: + logger.debug(f"Failed to fetch {path!r} from repository {repository.full_name!r}: {error}") + return None + + if isinstance(contents, list): + return None + + try: + return contents.decoded_content.decode("utf-8") + except (AssertionError, UnicodeDecodeError): + # Contents are not base64-encoded for files bigger than 1 MB, which `decoded_content` asserts + logger.debug(f"Could not decode {path!r} from repository {repository.full_name!r}") + return None + + +class GitConnector(GithubConnector): + """A class for connecting to a git repository hosted on GitHub.""" + _organization: str """The organization to which the repository belongs.""" @@ -402,21 +556,6 @@ def requirements_path(self) -> Path: """Path to the requirements.txt path of the repo, if present.""" return self.path / "requirements.txt" - @property - def authenticated(self) -> bool: - """Whether the current session is authenticated.""" - if self._connection is None: - return False - - from github import GithubException # noqa: PLC0415 - importing the GitHub API client is expensive - - try: - self._connection.get_user().login # noqa: B018 - login is a property - except GithubException: - return False - else: - return True - @property def worktrees_path(self) -> Path: """Path to the worktrees directory.""" @@ -430,59 +569,6 @@ def update(self): self.fetch() self.fetch_worktrees() - def connect(self): - """Connect to the Github API.""" - from github import Auth as GithubAuth, Github # noqa: PLC0415 - importing the GitHub API client is expensive - - if self._token is None: - - def get_token(prompt: bool) -> str | None: - return self.store.secrets.get( - GITHUB_DOMAIN, - scope="api", - fields=["password"], - prompt_format="GitHub API token:", - ask_missing=prompt, - ).password - - token = get_token(prompt=False) - - if not token: - logger.info( - """Connection to your GitHub account is necessary to pursue this operation, please configure a - Personal Access Token (classic) on https://github.com/settings/tokens with the following permissions: - - repo (all) - - user: - - read:user - - user:email - """ - ) - token = get_token(prompt=True) - - self._token = token - - if not self.connected: - self._connection = Github(auth=GithubAuth.Token(self._token)) # type: ignore [assignment] - - if not self.authenticated: - logger.warning("Failed to connect to Github API, please check your token is valid") - self.store.secrets.invalidate(GITHUB_DOMAIN, scope="api") - self._disconnect() - self.connect() - return - - logger.debug("Connected to Github API") - - def disconnect(self): - """Disconnect from the Github API.""" - self._disconnect() - logger.debug("Disconnected from Github API") - - def _disconnect(self): - """Disconnect from the Github API.""" - self._token = None - del self._connection - def _check_repository(self, force_clone: bool = False): """Check whether the repository exists locally.""" if self.repository is None: diff --git a/odev/common/odev.py b/odev/common/odev.py index 4d02b460b..e05cfcb48 100644 --- a/odev/common/odev.py +++ b/odev/common/odev.py @@ -1,5 +1,6 @@ """Self update Odev by pulling latest changes from the git repository.""" +import ast import contextlib import importlib import inspect @@ -61,7 +62,7 @@ UTC = timezone.utc -__all__ = ["Odev"] +__all__ = ["Odev", "parse_plugin_manifest", "plugin_module_name"] PRUNING_INTERVAL = 14 @@ -81,6 +82,9 @@ MIN_ARGV_LENGTH = 2 """Minimum number of command line arguments required (command and subcommand).""" +PLUGIN_MANIFEST_FILENAME = "__manifest__.py" +"""Name of the manifest file located at the root of a plugin repository.""" + class Manifest(TypedDict): """Plugin manifest information.""" @@ -102,6 +106,67 @@ class Plugin(NamedTuple): logger = logging.getLogger(__name__) +def plugin_module_name(plugin: str) -> str: + """Convert the name of a plugin to the name of the module it is linked to under the plugins directory. + + :param plugin: Name of the plugin, in the format `organization/repository` + :return: Name of the python module for this plugin + """ + return plugin.split("/")[-1].replace("-", "_") + + +def parse_plugin_manifest(source: str, name: str) -> Manifest | None: + """Extract the metadata of a plugin from the source of its manifest, without executing it. + + Only the module docstring and top-level assignments of literal values are read, which makes this function safe + to use on manifests originating from untrusted repositories. A source that does not define a top-level + `__version__` string is not considered a valid odev plugin manifest. + + :param source: Content of the `__manifest__.py` file + :param name: Name of the plugin, in the format `organization/repository` + :return: Manifest of the plugin, or `None` if the source is not a valid odev plugin manifest + """ + try: + tree = ast.parse(source) + except (SyntaxError, ValueError, RecursionError): + logger.debug(f"Failed to parse the manifest of plugin {name!r}") + return None + + assignments: dict[str, Any] = {} + + for node in tree.body: + if isinstance(node, ast.Assign): + targets, value = node.targets, node.value + elif isinstance(node, ast.AnnAssign) and node.value is not None: + targets, value = [node.target], node.value + else: + continue + + try: + literal = ast.literal_eval(value) + except (SyntaxError, TypeError, ValueError, MemoryError, RecursionError): + continue + + assignments.update({target.id: literal for target in targets if isinstance(target, ast.Name)}) + + manifest_version = assignments.get("__version__") + + if not isinstance(manifest_version, str): + logger.debug(f"Manifest of plugin {name!r} does not declare a version number") + return None + + depends = assignments.get("depends") + + return { + "name": name, + "version": manifest_version, + "description": (ast.get_docstring(tree) or "").strip(), + "depends": [dependency for dependency in depends if isinstance(dependency, str)] + if isinstance(depends, list) + else [], + } + + class Odev(Generic[CommandType]): """Main framework class.""" @@ -235,7 +300,7 @@ def dumps_path(self) -> Path: def plugins(self) -> Generator[Plugin, None, None]: """Yields enabled plugins sorted topologically.""" for plugin_name in self._plugins_dependency_tree(): - plugin_path = self.plugins_path / plugin_name.split("/")[-1].replace("-", "_") + plugin_path = self.plugins_path / plugin_module_name(plugin_name) plugin_manifest = self._load_plugin_manifest(plugin_path) yield Plugin(plugin_name, plugin_path, plugin_manifest) @@ -640,33 +705,33 @@ def _load_plugin_module(self, plugin: Plugin) -> None: :param plugin: Plugin whose module should be imported. """ # Module names MUST use underscores even if directories use dashes - plugin_module_name = plugin.path.name.replace("-", "_") - module_name = f"odev.plugins.{plugin_module_name}" + module_basename = plugin.path.name.replace("-", "_") + module_name = f"odev.plugins.{module_basename}" # Try to import directly from sys.path first try: - module = importlib.import_module(plugin_module_name) + module = importlib.import_module(module_basename) except ImportError: # Fallback to explicit file loading if direct import fails init_path = plugin.path / "__init__.py" if not init_path.exists(): return - spec = spec_from_file_location(plugin_module_name, init_path) + spec = spec_from_file_location(module_basename, init_path) if not spec or not spec.loader: return module = module_from_spec(spec) - sys.modules[plugin_module_name] = module + sys.modules[module_basename] = module try: spec.loader.exec_module(module) except Exception: # Drop the half-initialized module so a later retry starts from a clean state - sys.modules.pop(plugin_module_name, None) + sys.modules.pop(module_basename, None) raise # Ensure it's available as odev.plugins.X sys.modules[module_name] = module - setattr(sys.modules["odev.plugins"], plugin_module_name, module) + setattr(sys.modules["odev.plugins"], module_basename, module) def _install_missing_plugin_requirements(self) -> bool: """Install missing python packages from the requirements of all enabled plugins. @@ -788,7 +853,7 @@ def install_plugin(self, plugin: str, as_dependency: bool = False) -> None: for dependency in depends: self.install_plugin(dependency, as_dependency=True) - plugin_path = self.plugins_path / repository._repository.replace("-", "_") + plugin_path = self.plugins_path / plugin_module_name(repository.name) self.plugins_path.mkdir(parents=True, exist_ok=True) if self._plugin_is_installed(plugin): @@ -830,7 +895,7 @@ def uninstall_plugin(self, plugin: str) -> None: if installed_plugin == plugin: continue - installed_plugin_path = self.plugins_path / installed_plugin.split("/")[-1].replace("-", "_") + installed_plugin_path = self.plugins_path / plugin_module_name(installed_plugin) manifest = self._load_plugin_manifest(installed_plugin_path) if any(dep in manifest.get("depends", []) for dep in dependents | {plugin}): @@ -848,7 +913,7 @@ def uninstall_plugin(self, plugin: str) -> None: raise OdevError("Aborting plugin uninstallation") for dependent in dependents | {plugin}: - plugin_path = self.plugins_path / dependent.split("/")[-1].replace("-", "_") + plugin_path = self.plugins_path / plugin_module_name(dependent) plugin_path.unlink(missing_ok=True) self.config.plugins.enabled = {p for p in self.config.plugins.enabled if p != dependent} logger.info(f"Uninstalled plugin {dependent!r}") @@ -887,13 +952,15 @@ def _load_plugin_manifest(self, plugin_path: Path) -> Manifest: "depends": [], } - if not (plugin_path / "__manifest__.py").exists(): + manifest_path = plugin_path / PLUGIN_MANIFEST_FILENAME + + if not manifest_path.exists(): return defaults - spec = spec_from_file_location(f"{plugin_path.name}.__manifest__", (plugin_path / "__manifest__.py").as_posix()) + spec = spec_from_file_location(f"{plugin_path.name}.__manifest__", manifest_path.as_posix()) if spec is None: - raise ImportError(f"Cannot load manifest module from {(plugin_path / '__manifest__.py').as_posix()}") + raise ImportError(f"Cannot load manifest module from {manifest_path.as_posix()}") manifest = module_from_spec(spec) cast(Loader, spec.loader).exec_module(manifest) diff --git a/tests/tests/commands/test_utilities.py b/tests/tests/commands/test_utilities.py index 7974c2a77..2144aad0d 100644 --- a/tests/tests/commands/test_utilities.py +++ b/tests/tests/commands/test_utilities.py @@ -1,5 +1,8 @@ +import os import shutil from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch from odev._version import __version__ from odev.common.python import PythonEnv @@ -9,6 +12,7 @@ POSTGRES_PATH = "odev.common.connectors.PostgresConnector" GIT_PATH = "odev.common.connectors.git.GitConnector" +GITHUB_PATH = "odev.common.connectors.git.GithubConnector" class TestCommandUtilities(OdevCommandTestCase): @@ -240,6 +244,221 @@ def test_plugin_03_enable_dependencies(self): self.assertFalse(self.odev._plugin_is_installed(plugin)) self.assertFalse(self.odev._plugin_is_installed(dependent)) + def test_plugin_04_list(self): + """Run the command to list plugins, showing enabled and downloaded ones alike.""" + self.__enable_test_plugin() + stdout, _ = self.__dispatch_plugin("--list") + + self.assertRegex(stdout, r"Plugin\s+Version\s+Branch\s+State\s+Depends") + self.assertRegex(stdout, r"test/test-plugin\s+1\.0\.0\s+enabled") + self.assertRegex(stdout, r"test/test-plugin-dep\s+1\.0\.0\s+disabled\s+test-plugin") + self.assertNotIn("test/test-addons", stdout) + + def test_plugin_05_list_shadowed(self): + """Run the command to list plugins when two enabled plugins share the same module name.""" + self.__enable_test_plugin() + self.odev.config.plugins.enabled = [*self.odev.config.plugins.enabled, "other/test-plugin"] + + stdout, _ = self.__dispatch_plugin("--list") + + self.assertRegex(stdout, r"test/test-plugin\s+1\.0\.0\s+enabled") + self.assertRegex(stdout, r"other/test-plugin\s+shadowed") + self.assertIn("other/test-plugin (shadowed by test/test-plugin)", stdout) + + def test_plugin_06_search(self): + """Run the command to search plugins on GitHub, keeping only valid plugin repositories.""" + self.odev.config.paths.repositories = self.res_path / "repositories" + manifest = (self.res_path / "repositories" / "test" / "test-plugin" / "__manifest__.py").read_text() + repositories = [ + self.__github_repository("test/archived-plugin", archived=True), + self.__github_repository("odoo-odev/odev-plugin-template"), + self.__github_repository("test/test-plugin", stars=3), + self.__github_repository("other/remote-plugin"), + self.__github_repository("other/not-a-plugin", description="Not a plugin at all"), + ] + + with ( + self.patch(GITHUB_PATH, "search_repositories", return_value=repositories) as search, + self.patch(GITHUB_PATH, "get_repository_file", side_effect=[manifest, manifest, None]), + ): + stdout, _ = self.__dispatch_plugin("--search") + + self.assertRegex(stdout, r"Plugin\s+Version\s+Stars\s+State\s+Description") + self.assertRegex(stdout, r"test/test-plugin\s+1\.0\.0\s+3\s+disabled") + self.assertRegex(stdout, r"other/remote-plugin\s+1\.0\.0\s+0\s+not downloaded") + self.assertNotIn("other/not-a-plugin", stdout) + self.assertNotIn("test/archived-plugin", stdout) + self.assertNotIn("odev-plugin-template", stdout) + self.assertTrue(search.call_args.args[0].startswith("odev plugin")) + + def test_plugin_07_search_terms(self): + """Run the command to search plugins with additional terms, showing the state of enabled plugins.""" + self.__enable_test_plugin() + manifest = (self.res_path / "repositories" / "test" / "test-plugin" / "__manifest__.py").read_text() + + with ( + self.patch( + GITHUB_PATH, + "search_repositories", + return_value=[self.__github_repository("test/test-plugin")], + ) as search, + self.patch(GITHUB_PATH, "get_repository_file", return_value=manifest), + ): + stdout, _ = self.__dispatch_plugin("--search", "editor") + + self.assertRegex(stdout, r"test/test-plugin\s+1\.0\.0\s+0\s+enabled") + self.assertIn("editor", search.call_args.args[0]) + + def test_plugin_08_search_no_result(self): + """Run the command to search plugins when no repository exposes a valid manifest.""" + self.odev.config.paths.repositories = self.res_path / "repositories" + + with ( + self.patch( + GITHUB_PATH, + "search_repositories", + return_value=[self.__github_repository("other/not-a-plugin")], + ), + self.patch(GITHUB_PATH, "get_repository_file", return_value=None), + ): + _, stderr = self.dispatch_command("plugin", "--search", "unknown") + + self.assertIn("No odev plugin found matching 'unknown'", stderr) + + def test_plugin_09_show_qualified_name(self): + """Run the command to show a plugin using its fully qualified name.""" + self.__enable_test_plugin() + + stdout, _ = self.dispatch_command("plugin", "--show", "test/test-plugin") + self.assertIn("Plugin 'test/test-plugin' is enabled", stdout) + + stdout, _ = self.dispatch_command("plugin", "--show", "test-plugin") + self.assertIn("Plugin 'test/test-plugin' is enabled", stdout) + + def test_plugin_10_show_downloaded(self): + """Run the command to show a plugin available locally but not enabled, without querying GitHub.""" + self.__enable_test_plugin() + + with self.patch(GITHUB_PATH, "get_repository", return_value=None) as get_repository: + stdout, _ = self.__dispatch_plugin("--show", "test/test-plugin-dep") + + self.assertIn("Plugin 'test/test-plugin-dep' is disabled", stdout) + self.assertRegex(stdout, r"Version:\s+1\.0\.0") + self.assertRegex(stdout, r"Depends:\s+test/test-plugin") + self.assertRegex(stdout, r"Path:\s+.*test-plugin-dep") + self.assertRegex(stdout, r"URL:\s+https://github\.com/test/test-plugin-dep") + get_repository.assert_not_called() + + def test_plugin_11_show_not_downloaded(self): + """Run the command to show a plugin that is not available locally, fetching its manifest from GitHub.""" + self.odev.config.paths.repositories = self.res_path / "repositories" + manifest = (self.res_path / "repositories" / "test" / "test-plugin" / "__manifest__.py").read_text() + + with ( + self.patch( + GITHUB_PATH, + "get_repository", + return_value=self.__github_repository("other/remote-plugin", stars=42), + ) as get_repository, + self.patch(GITHUB_PATH, "get_repository_file", return_value=manifest) as get_repository_file, + ): + stdout, _ = self.__dispatch_plugin("--show", "other/remote-plugin") + + self.assertIn("Plugin 'other/remote-plugin' is not downloaded", stdout) + self.assertRegex(stdout, r"Version:\s+1\.0\.0") + self.assertRegex(stdout, r"Branch:\s+main") + self.assertRegex(stdout, r"Stars:\s+42") + self.assertNotIn("Path:", stdout) + self.assertIn("Run 'odev plugin --enable other/remote-plugin' to install this plugin", stdout) + get_repository.assert_called_once_with("other/remote-plugin") + self.assertEqual(get_repository_file.call_args.args[1], "__manifest__.py") + + def test_plugin_12_show_archived_and_excluded(self): + """Run the command to show plugins that can be found on GitHub but should not be installed.""" + self.odev.config.paths.repositories = self.res_path / "repositories" + manifest = (self.res_path / "repositories" / "test" / "test-plugin" / "__manifest__.py").read_text() + + with ( + self.patch( + GITHUB_PATH, + "get_repository", + return_value=self.__github_repository("other/archived-plugin", archived=True), + ), + self.patch(GITHUB_PATH, "get_repository_file", return_value=manifest), + ): + stdout, _ = self.__dispatch_plugin("--show", "other/archived-plugin") + + self.assertIn("Repository 'other/archived-plugin' is archived", stdout) + + with ( + self.patch( + GITHUB_PATH, + "get_repository", + return_value=self.__github_repository("odoo-odev/odev-plugin-template"), + ), + self.patch(GITHUB_PATH, "get_repository_file", return_value=manifest), + ): + stdout, _ = self.__dispatch_plugin("--show", "odoo-odev/odev-plugin-template") + + self.assertIn("is a template used to create new plugins and cannot be installed", stdout) + self.assertNotIn("--enable odoo-odev/odev-plugin-template", stdout) + + def test_plugin_13_show_unknown(self): + """Run the command to show a plugin that is neither available locally nor on GitHub.""" + self.odev.config.paths.repositories = self.res_path / "repositories" + + with self.patch(GITHUB_PATH, "get_repository", return_value=None): + stdout, _ = self.__dispatch_plugin("--show", "other/unknown-plugin") + + self.assertIn("Plugin 'other/unknown-plugin' is not downloaded", stdout) + self.assertNotIn("Version:", stdout) + + with self.patch(GITHUB_PATH, "get_repository", return_value=None) as get_repository: + stdout, _ = self.__dispatch_plugin("--show", "unknown-plugin") + + self.assertIn("Plugin 'unknown-plugin' is not downloaded", stdout) + self.assertIn("Use the full name of the plugin", stdout) + get_repository.assert_not_called() + + def test_plugin_14_show_all(self): + """Run the command to show all plugins available locally, enabled or not.""" + self.__enable_test_plugin() + + with self.patch(GITHUB_PATH, "get_repository", return_value=None) as get_repository: + stdout, _ = self.__dispatch_plugin("--show") + + self.assertIn("Plugin 'test/test-plugin' is enabled", stdout) + self.assertIn("Plugin 'test/test-plugin-dep' is disabled", stdout) + get_repository.assert_not_called() + + def __dispatch_plugin(self, *arguments: str) -> tuple[str, str]: + """Run the plugin command on a wide terminal so table columns are not cropped.""" + with patch.dict(os.environ, {"COLUMNS": "200"}): + return self.dispatch_command("plugin", *arguments) + + def __enable_test_plugin(self): + """Enable the test plugin from the local test resources and unlink it after the test.""" + self.odev.config.paths.repositories = self.res_path / "repositories" + plugin_link = Path(self.odev.plugins_path) / "test_plugin" + self.addCleanup(plugin_link.unlink, missing_ok=True) + + with ( + self.patch_property(GIT_PATH, "exists", value=True), + self.patch(GIT_PATH, "update"), + ): + self.dispatch_command("plugin", "--enable", "test/test-plugin") + + def __github_repository(self, full_name: str, description: str = "", stars: int = 0, archived: bool = False): + """Build a stand-in for a repository as returned by the GitHub API.""" + return SimpleNamespace( + full_name=full_name, + html_url=f"https://github.com/{full_name}", + description=description, + stargazers_count=stars, + archived=archived, + default_branch="main", + ) + class TestCommandUtilitiesVenv(OdevCommandTestCase): @classmethod diff --git a/tests/tests/common/test_git_connector.py b/tests/tests/common/test_git_connector.py index f91e7b9fc..511529b8b 100644 --- a/tests/tests/common/test_git_connector.py +++ b/tests/tests/common/test_git_connector.py @@ -1,4 +1,8 @@ -from odev.common.connectors.git import GitConnector +from types import SimpleNamespace + +from github import GithubException, UnknownObjectException + +from odev.common.connectors.git import GitConnector, GithubConnector from odev.common.errors import ConnectorError from tests.fixtures import OdevTestCase @@ -25,3 +29,36 @@ def test_invalid_repo_format_raises(self): with self.assertRaises(ConnectorError) as ctx: GitConnector("onlyonepart") self.assertIn("Invalid repository format", str(ctx.exception)) + + +class TestGithubConnectorRepositories(OdevTestCase): + def __connector(self, get_repo) -> GithubConnector: + """Build a connector already connected to a stand-in of the Github API.""" + connector = GithubConnector() + connector._connection = SimpleNamespace(get_repo=get_repo) # type: ignore [assignment] + + for attribute in ("connect", "disconnect"): + patcher = self.patch(connector, attribute) + patcher.start() + self.addCleanup(patcher.stop) + + return connector + + def test_get_repository(self): + repository = SimpleNamespace(full_name="acme/myrepo") + connector = self.__connector(lambda name: repository) + self.assertIs(connector.get_repository("acme/myrepo"), repository) + + def test_get_repository_missing(self): + def get_repo(name: str): + raise UnknownObjectException(404, None, None) + + connector = self.__connector(get_repo) + self.assertIsNone(connector.get_repository("acme/missing")) + + def test_get_repository_error(self): + def get_repo(name: str): + raise GithubException(500, None, None) + + connector = self.__connector(get_repo) + self.assertIsNone(connector.get_repository("acme/myrepo")) diff --git a/tests/tests/common/test_odev.py b/tests/tests/common/test_odev.py index 0d24ec0c8..9f3004ca0 100644 --- a/tests/tests/common/test_odev.py +++ b/tests/tests/common/test_odev.py @@ -6,7 +6,7 @@ from odev._version import __version__ from odev.common.commands import Command -from odev.common.odev import Manifest, Plugin, logger +from odev.common.odev import Manifest, Plugin, logger, parse_plugin_manifest, plugin_module_name from tests.fixtures import CaptureOutput, OdevTestCase @@ -262,3 +262,55 @@ def test_20_load_plugins_repoints_preexisting_plugins_module(self): self.odev.load_plugins() self.assertEqual(sys.modules["odev.plugins"].__path__, [str(self.odev.plugins_path)]) + + def test_21_plugin_module_name(self): + """The module name of a plugin should drop the organization and use underscores.""" + self.assertEqual(plugin_module_name("odoo-odev/odev-plugin-editor-base"), "odev_plugin_editor_base") + self.assertEqual(plugin_module_name("odev-plugin-ai"), "odev_plugin_ai") + + def test_22_parse_plugin_manifest(self): + """The manifest of a plugin should be parsed into its name, version, description and dependencies.""" + manifest = parse_plugin_manifest( + '"""Some plugin."""\n\n__version__ = "1.2.3"\n\ndepends = ["test/test-plugin", 42]\n', + "test/test-plugin-dep", + ) + + self.assertEqual( + manifest, + { + "name": "test/test-plugin-dep", + "version": "1.2.3", + "description": "Some plugin.", + "depends": ["test/test-plugin"], + }, + ) + + def test_23_parse_plugin_manifest_invalid(self): + """A source that is not a valid plugin manifest should be rejected.""" + self.assertIsNone(parse_plugin_manifest('"""No version."""\n\ndepends = []\n', "test/test-plugin")) + self.assertIsNone(parse_plugin_manifest("def invalid(:\n", "test/test-plugin")) + self.assertIsNone(parse_plugin_manifest("__version__ = 1.0\n", "test/test-plugin")) + self.assertIsNone(parse_plugin_manifest('{"name": "Sales", "version": "17.0"}\n', "test/test-addons")) + + def test_24_parse_plugin_manifest_does_not_execute_code(self): + """Parsing the manifest of an untrusted repository should never execute its content.""" + with self.patch("odev.common.odev.logger", "warning") as logger_warning: + manifest = parse_plugin_manifest( + '"""Malicious plugin."""\n' + "import odev.common.odev as target\n" + 'target.logger.warning("executed")\n' + "raise SystemExit(1)\n" + '__version__ = "6.6.6"\n', + "evil/plugin", + ) + + self.assertEqual( + manifest, + { + "name": "evil/plugin", + "version": "6.6.6", + "description": "Malicious plugin.", + "depends": [], + }, + ) + logger_warning.assert_not_called()