Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/plugins/install.py: 100%
83 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 23:36 +0000
1# SPDX-License-Identifier: MIT
2# Copyright © 2025 Bijan Mousavi
4"""Implements the `plugins install` subcommand for the Bijux CLI.
6This module contains the logic for installing a new plugin by copying its
7source directory into the CLI's plugins folder. The process is designed to be
8atomic and safe, incorporating validation of the plugin's name and metadata,
9version compatibility checks against the current CLI version, and file locking
10to prevent race conditions during installation.
12Output Contract:
13 * Install Success: `{"status": "installed", "plugin": str, "dest": str}`
14 * Dry Run Success: `{"status": "dry-run", "plugin": str, ...}`
15 * Error: `{"error": "...", "code": int}`
17Exit Codes:
18 * `0`: Success.
19 * `1`: A fatal error occurred (e.g., source not found, invalid name,
20 version incompatibility, filesystem error).
21 * `2`: An invalid flag was provided (e.g., bad format).
22 * `3`: An ASCII or encoding error was detected in the environment.
23"""
25from __future__ import annotations
27from collections.abc import Iterator
28import contextlib
29import errno
30import fcntl
31from pathlib import Path
32import shutil
33import tempfile
35import typer
37from bijux_cli.commands.plugins.utils import (
38 PLUGIN_NAME_RE,
39 ignore_hidden_and_broken_symlinks,
40 parse_required_cli_version,
41 refuse_on_symlink,
42)
43from bijux_cli.commands.utilities import (
44 emit_error_and_exit,
45 new_run_command,
46 validate_common_flags,
47)
48from bijux_cli.core.constants import (
49 HELP_DEBUG,
50 HELP_FORMAT,
51 HELP_NO_PRETTY,
52 HELP_QUIET,
53 HELP_VERBOSE,
54)
55from bijux_cli.services.plugins import get_plugins_dir
58def install_plugin(
59 path: str = typer.Argument(..., help="Path to plugin directory"),
60 dry_run: bool = typer.Option(False, "--dry-run"),
61 force: bool = typer.Option(False, "--force", "-F"),
62 quiet: bool = typer.Option(False, "-q", "--quiet", help=HELP_QUIET),
63 verbose: bool = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE),
64 fmt: str = typer.Option("json", "-f", "--format", help=HELP_FORMAT),
65 pretty: bool = typer.Option(False, "--pretty/--no-pretty", help=HELP_NO_PRETTY),
66 debug: bool = typer.Option(False, "-d", "--debug", help=HELP_DEBUG),
67) -> None:
68 """Installs a plugin from a local source directory.
70 This function orchestrates the plugin installation process. It validates
71 the source path and plugin name, checks for version compatibility, handles
72 pre-existing plugins via the `--force` flag, and performs an atomic copy
73 into the plugins directory using a file lock and temporary directory.
75 Args:
76 path (str): The source path to the plugin directory to install.
77 dry_run (bool): If True, simulates the installation without making changes.
78 force (bool): If True, overwrites an existing plugin of the same name.
79 quiet (bool): If True, suppresses all output except for errors.
80 verbose (bool): If True, includes runtime metadata in error payloads.
81 fmt (str): The output format for confirmation or error messages.
82 pretty (bool): If True, pretty-prints the output.
83 debug (bool): If True, enables debug diagnostics.
85 Returns:
86 None:
88 Raises:
89 SystemExit: Always exits with a contract-compliant status code and
90 payload, indicating success or detailing an error.
91 """
92 from packaging.specifiers import SpecifierSet
94 from bijux_cli.__version__ import version as cli_version
96 command = "plugins install"
98 fmt_lower = validate_common_flags(fmt, command, quiet)
99 plugins_dir = get_plugins_dir()
100 refuse_on_symlink(plugins_dir, command, fmt_lower, quiet, verbose, debug)
102 src = Path(path).expanduser()
103 try:
104 src = src.resolve()
105 except (FileNotFoundError, OSError, RuntimeError):
106 src = src.absolute()
107 if not src.exists() or not src.is_dir():
108 emit_error_and_exit(
109 "Source not found",
110 code=1,
111 failure="source_not_found",
112 command=command,
113 fmt=fmt_lower,
114 quiet=quiet,
115 include_runtime=verbose,
116 debug=debug,
117 )
119 plugin_name = src.name
121 if not PLUGIN_NAME_RE.fullmatch(plugin_name) or not plugin_name.isascii():
122 emit_error_and_exit(
123 "Invalid plugin name: only ASCII letters, digits, dash and underscore are allowed.",
124 code=1,
125 failure="invalid_name",
126 command=command,
127 fmt=fmt_lower,
128 quiet=quiet,
129 include_runtime=verbose,
130 debug=debug,
131 )
133 dest = plugins_dir / plugin_name
135 try:
136 plugins_dir.mkdir(parents=True, exist_ok=True)
137 except Exception as exc:
138 emit_error_and_exit(
139 f"Cannot create plugins dir '{plugins_dir}': {exc}",
140 code=1,
141 failure="create_dir_failed",
142 command=command,
143 fmt=fmt_lower,
144 quiet=quiet,
145 include_runtime=verbose,
146 debug=debug,
147 )
149 lock_file = plugins_dir / ".bijux_install.lock"
151 @contextlib.contextmanager
152 def _lock(fp: Path) -> Iterator[None]:
153 """Provides an exclusive, non-blocking file lock.
155 This context manager attempts to acquire a lock on the specified file.
156 It is used to ensure atomic filesystem operations within the plugins
157 directory.
159 Args:
160 fp (Path): The path to the file to lock.
162 Yields:
163 None: Yields control to the `with` block once the lock is acquired.
164 """
165 with fp.open("w") as fh:
166 fcntl.flock(fh, fcntl.LOCK_EX)
167 try:
168 yield
169 finally:
170 fcntl.flock(fh, fcntl.LOCK_UN)
172 with _lock(lock_file):
173 if plugins_dir.is_symlink():
174 emit_error_and_exit(
175 f"Refusing to install: plugins dir '{plugins_dir}' is a symlink.",
176 code=1,
177 failure="symlink_dir",
178 command=command,
179 fmt=fmt_lower,
180 quiet=quiet,
181 include_runtime=verbose,
182 debug=debug,
183 )
185 if dest.exists():
186 if not force:
187 emit_error_and_exit(
188 f"Plugin '{plugin_name}' already installed. Use --force.",
189 code=1,
190 failure="already_installed",
191 command=command,
192 fmt=fmt_lower,
193 quiet=quiet,
194 include_runtime=verbose,
195 debug=debug,
196 )
197 try:
198 if dest.is_dir():
199 shutil.rmtree(dest)
200 else:
201 dest.unlink()
202 except Exception as exc:
203 emit_error_and_exit(
204 f"Unable to remove existing '{dest}': {exc}",
205 code=1,
206 failure="remove_failed",
207 command=command,
208 fmt=fmt_lower,
209 quiet=quiet,
210 include_runtime=verbose,
211 debug=debug,
212 )
214 plugin_py = src / "plugin.py"
215 if not plugin_py.exists():
216 emit_error_and_exit(
217 "plugin.py not found in plugin directory",
218 code=1,
219 failure="plugin_py_missing",
220 command=command,
221 fmt=fmt_lower,
222 quiet=quiet,
223 include_runtime=verbose,
224 debug=debug,
225 )
226 version_spec = parse_required_cli_version(plugin_py)
227 if version_spec:
228 try:
229 spec = SpecifierSet(version_spec)
230 if not spec.contains(cli_version, prereleases=True):
231 emit_error_and_exit(
232 f"Incompatible CLI version: plugin requires '{version_spec}', but you have '{cli_version}'",
233 code=1,
234 failure="incompatible_version",
235 command=command,
236 fmt=fmt_lower,
237 quiet=quiet,
238 include_runtime=verbose,
239 debug=debug,
240 )
241 except Exception as exc:
242 emit_error_and_exit(
243 f"Invalid version specifier in plugin: '{version_spec}'. {exc}",
244 code=1,
245 failure="invalid_specifier",
246 command=command,
247 fmt=fmt_lower,
248 quiet=quiet,
249 include_runtime=verbose,
250 debug=debug,
251 )
253 if dry_run:
254 payload = {
255 "status": "dry-run",
256 "plugin": plugin_name,
257 "source": str(src),
258 "dest": str(dest),
259 }
260 else:
261 with tempfile.TemporaryDirectory(dir=plugins_dir) as td:
262 tmp_dst = Path(td) / plugin_name
263 try:
264 shutil.copytree(
265 src,
266 tmp_dst,
267 symlinks=True,
268 ignore=ignore_hidden_and_broken_symlinks,
269 )
270 except OSError as exc:
271 if exc.errno == errno.ENOSPC or "No space left on device" in str(
272 exc
273 ):
274 emit_error_and_exit(
275 "Disk full during plugin install",
276 code=1,
277 failure="disk_full",
278 command=command,
279 fmt=fmt_lower,
280 quiet=quiet,
281 include_runtime=verbose,
282 debug=debug,
283 )
284 if exc.errno == errno.EACCES or "Permission denied" in str(exc):
285 emit_error_and_exit(
286 "Permission denied during plugin install",
287 code=1,
288 failure="permission_denied",
289 command=command,
290 fmt=fmt_lower,
291 quiet=quiet,
292 include_runtime=verbose,
293 debug=debug,
294 )
295 emit_error_and_exit(
296 f"OSError during plugin install: {exc!r}",
297 code=1,
298 failure="os_error",
299 command=command,
300 fmt=fmt_lower,
301 quiet=quiet,
302 include_runtime=verbose,
303 debug=debug,
304 )
305 if not (tmp_dst / "plugin.py").is_file():
306 emit_error_and_exit(
307 f"plugin.py not found in '{tmp_dst}'",
308 code=1,
309 failure="plugin_py_missing_after_copy",
310 command=command,
311 fmt=fmt_lower,
312 quiet=quiet,
313 include_runtime=verbose,
314 debug=debug,
315 )
316 shutil.move(str(tmp_dst), dest)
317 payload = {"status": "installed", "plugin": plugin_name, "dest": str(dest)}
319 new_run_command(
320 command_name=command,
321 payload_builder=lambda include: payload,
322 quiet=quiet,
323 verbose=verbose,
324 fmt=fmt_lower,
325 pretty=pretty,
326 debug=debug,
327 )