Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / plugins / commands / uninstall.py: 97%
74 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-01-26 17:59 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright © 2025 Bijan Mousavi
4"""Implements the `plugins uninstall` subcommand for the Bijux CLI.
6This module contains the logic for permanently removing an installed plugin
7from the filesystem. The operation locates the plugin directory by its exact
8name, performs security checks (e.g., refusing to act on symbolic links),
9and uses a file lock to ensure atomicity before deleting the directory.
11Output Contract:
12 * Success: `{"status": "uninstalled", "plugin": str}`
13 * Error: `{"error": str, "code": int}`
15Exit Codes:
16 * `0`: Success.
17 * `1`: A fatal error occurred (e.g., plugin not found, permission denied,
18 filesystem error).
19 * `2`: An invalid flag was provided (e.g., bad format).
20 * `3`: An ASCII or encoding error was detected in the environment.
21"""
23from __future__ import annotations
25from collections.abc import Iterator
26import contextlib
27import fcntl
28from pathlib import Path
29import shutil
30import subprocess # noqa: S603 # nosec B404 - intentional CLI invocation
31import sys
32import unicodedata
34import typer
36from bijux_cli.cli.core.command import (
37 new_run_command,
38 raise_exit_intent,
39 validate_common_flags,
40)
41from bijux_cli.cli.core.constants import (
42 OPT_FORMAT,
43 OPT_LOG_LEVEL,
44 OPT_PRETTY,
45 OPT_QUIET,
46)
47from bijux_cli.cli.core.help_text import (
48 HELP_FORMAT,
49 HELP_LOG_LEVEL,
50 HELP_NO_PRETTY,
51 HELP_QUIET,
52)
53from bijux_cli.cli.plugins.commands.validation import refuse_on_symlink
54from bijux_cli.core.enums import ErrorType
55from bijux_cli.core.precedence import current_execution_policy
56from bijux_cli.plugins import get_plugins_dir
57from bijux_cli.plugins.metadata import get_plugin_metadata, invalidate_plugin_cache
60def uninstall_plugin(
61 name: str = typer.Argument(..., help="Plugin name"),
62 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET),
63 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT),
64 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY),
65 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL),
66) -> None:
67 """Removes an installed plugin by deleting its directory.
69 This function locates the plugin directory by name, performs several safety
70 checks, acquires a file lock to ensure atomicity, and then permanently
71 removes the plugin from the filesystem.
73 Args:
74 name (str): The name of the plugin to uninstall. The match is
75 case-sensitive and Unicode-aware.
76 quiet (bool): If True, suppresses all output except for errors.
77 fmt (str): The output format for confirmation or error messages.
78 pretty (bool): If True, pretty-prints the output. log_level (str): Logging level for diagnostics.
80 Returns:
81 None:
83 Raises:
84 SystemExit: Always exits with a contract-compliant status code and
85 payload, indicating success or detailing an error.
86 """
87 command = "plugins uninstall"
89 policy = current_execution_policy()
90 quiet = policy.quiet
91 include_runtime = policy.include_runtime
92 log_level_value = policy.log_level
93 pretty = policy.pretty
94 fmt_lower = validate_common_flags(
95 fmt,
96 command,
97 quiet,
98 include_runtime=include_runtime,
99 log_level=log_level_value,
100 )
101 try:
102 meta = get_plugin_metadata(name)
103 except Exception:
104 meta = None
106 if meta and meta.source == "entrypoint" and meta.dist_name:
107 cmd = [sys.executable, "-m", "pip", "uninstall", "-y", meta.dist_name]
108 proc = subprocess.run( # noqa: S603 # nosec B603 - controlled command list
109 cmd,
110 capture_output=True,
111 text=True,
112 )
113 if proc.returncode != 0: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true
114 detail = proc.stderr.strip() or proc.stdout.strip()
115 raise_exit_intent(
116 f"pip uninstall failed: {detail}",
117 code=1,
118 failure="pip_uninstall_failed",
119 error_type=ErrorType.PLUGIN,
120 command=command,
121 fmt=fmt_lower,
122 quiet=quiet,
123 include_runtime=include_runtime,
124 log_level=log_level_value,
125 )
126 invalidate_plugin_cache()
127 payload = {"status": "uninstalled", "plugin": name}
128 new_run_command(
129 command_name=command,
130 payload_builder=lambda include: payload,
131 quiet=quiet,
132 fmt=fmt_lower,
133 pretty=pretty,
134 log_level=log_level_value,
135 )
137 plugins_dir = get_plugins_dir()
138 refuse_on_symlink(plugins_dir, command, fmt_lower, quiet, log_level_value)
140 lock_file = plugins_dir / ".bijux_install.lock"
142 plugin_dirs: list[Path] = []
143 try:
144 plugin_dirs = [
145 p
146 for p in plugins_dir.iterdir()
147 if p.is_dir()
148 and unicodedata.normalize("NFC", p.name)
149 == unicodedata.normalize("NFC", name)
150 ]
151 except Exception as exc:
152 raise_exit_intent(
153 f"Could not list plugins dir '{plugins_dir}': {exc}",
154 code=1,
155 failure="list_failed",
156 error_type=ErrorType.PLUGIN,
157 command=command,
158 fmt=fmt_lower,
159 quiet=quiet,
160 include_runtime=include_runtime,
161 log_level=log_level_value,
162 )
164 if not plugin_dirs:
165 raise_exit_intent(
166 f"Plugin '{name}' is not installed.",
167 code=1,
168 failure="not_installed",
169 error_type=ErrorType.PLUGIN,
170 command=command,
171 fmt=fmt_lower,
172 quiet=quiet,
173 include_runtime=include_runtime,
174 log_level=log_level_value,
175 )
177 plug_path = plugin_dirs[0]
179 @contextlib.contextmanager
180 def _lock(fp: Path) -> Iterator[None]:
181 """Provides an exclusive, non-blocking file lock.
183 This context manager attempts to acquire a lock on the specified file.
184 It is used to ensure atomic filesystem operations within the plugins
185 directory.
187 Args:
188 fp (Path): The path to the file to lock.
190 Yields:
191 None: Yields control to the `with` block once the lock is acquired.
192 """
193 fp.parent.mkdir(parents=True, exist_ok=True)
194 with fp.open("w") as fh:
195 fcntl.flock(fh, fcntl.LOCK_EX)
196 try:
197 yield
198 finally:
199 fcntl.flock(fh, fcntl.LOCK_UN)
201 with _lock(lock_file):
202 if not plug_path.exists():
203 pass
204 elif plug_path.is_symlink():
205 raise_exit_intent(
206 f"Plugin path '{plug_path}' is a symlink. Refusing to uninstall.",
207 code=1,
208 failure="symlink_path",
209 error_type=ErrorType.PLUGIN,
210 command=command,
211 fmt=fmt_lower,
212 quiet=quiet,
213 include_runtime=include_runtime,
214 log_level=log_level_value,
215 )
216 elif not plug_path.is_dir():
217 raise_exit_intent(
218 f"Plugin path '{plug_path}' is not a directory.",
219 code=1,
220 failure="not_dir",
221 error_type=ErrorType.PLUGIN,
222 command=command,
223 fmt=fmt_lower,
224 quiet=quiet,
225 include_runtime=include_runtime,
226 log_level=log_level_value,
227 )
228 else:
229 try:
230 shutil.rmtree(plug_path)
231 except PermissionError:
232 raise_exit_intent(
233 f"Permission denied removing '{plug_path}'",
234 code=1,
235 failure="permission_denied",
236 error_type=ErrorType.PLUGIN,
237 command=command,
238 fmt=fmt_lower,
239 quiet=quiet,
240 include_runtime=include_runtime,
241 log_level=log_level_value,
242 )
243 except Exception as exc:
244 raise_exit_intent(
245 f"Failed to remove '{plug_path}': {exc}",
246 code=1,
247 failure="remove_failed",
248 error_type=ErrorType.PLUGIN,
249 command=command,
250 fmt=fmt_lower,
251 quiet=quiet,
252 include_runtime=include_runtime,
253 log_level=log_level_value,
254 )
256 invalidate_plugin_cache()
257 payload = {"status": "uninstalled", "plugin": name}
259 new_run_command(
260 command_name=command,
261 payload_builder=lambda include: payload,
262 quiet=quiet,
263 fmt=fmt_lower,
264 pretty=pretty,
265 log_level=log_level,
266 )