diff --git a/CLAUDE.md b/CLAUDE.md index 71c99e7..783887a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -52,6 +52,7 @@ devsync/ │ ├── mcp_credential_prompter.py # MCP credential prompting │ ├── repository.py # Parse ai-config-kit.yaml │ ├── git_operations.py # Git clone/pull operations +│ ├── pip_utils.py # Pip package validation, detection, installation │ ├── checksum.py # File integrity checking │ └── conflict_resolution.py # Handle file conflicts ├── llm/ # LLM provider abstraction (HTTP-only, no SDK deps) @@ -355,6 +356,7 @@ devsync install https://github.com/company/standards devsync install ./package --tool claude --tool cursor devsync install ./package --no-ai devsync install ./package --conflict skip +devsync install ./package --skip-pip # List installed packages devsync list diff --git a/README.md b/README.md index 70f839b..ccf857e 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ No API key? DevSync works without one -- it falls back to file-copy mode. Add `- - **AI-powered extraction** -- LLM reads your project's rules, MCP configs, and commands to produce abstract practice declarations - **AI-powered installation** -- LLM adapts incoming practices to your existing setup with intelligent merging - **23+ AI tool integrations** -- Claude Code, Cursor, Windsurf, GitHub Copilot, Kiro, Roo Code, Cline, Codex, and more +- **MCP server dependencies** -- auto-detects pip-installable MCP servers and prompts to install them (`--skip-pip` to skip) - **MCP credential handling** -- prompts for credentials at install time, never stores them in repos - **v1 backward compatibility** -- old `ai-config-kit-package.yaml` packages still install via file-copy - **Graceful degradation** -- works without an API key, `--no-ai` flag for explicit file-copy mode diff --git a/devsync/cli/install_v2.py b/devsync/cli/install_v2.py index a9b5252..529230c 100644 --- a/devsync/cli/install_v2.py +++ b/devsync/cli/install_v2.py @@ -1,9 +1,14 @@ """V2 install command — AI-powered package installation.""" +from __future__ import annotations + import shutil import tempfile from pathlib import Path -from typing import Optional +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from devsync.core.practice import MCPDeclaration from rich.console import Console from rich.prompt import Confirm @@ -26,6 +31,7 @@ def install_v2_command( no_ai: bool = False, conflict: str = "prompt", project_dir: Optional[str] = None, + skip_pip: bool = False, ) -> int: """Install a package into the current project. @@ -37,6 +43,7 @@ def install_v2_command( no_ai: Disable AI-powered adaptation. conflict: Conflict strategy ('prompt', 'skip', 'overwrite', 'rename'). project_dir: Target project directory. Defaults to cwd. + skip_pip: Skip pip package installations for MCP servers. Returns: Exit code (0 = success). @@ -72,8 +79,8 @@ def install_v2_command( console.print(f" Tools: {', '.join(target_tools)}") if manifest.is_v2 and manifest.has_practices and not no_ai: - return _install_v2_ai(manifest, project_root, target_tools) - return _install_v2_fallback(manifest, package_path, project_root, target_tools, conflict) + return _install_v2_ai(manifest, project_root, target_tools, skip_pip=skip_pip) + return _install_v2_fallback(manifest, package_path, project_root, target_tools, conflict, skip_pip=skip_pip) finally: if cloned_tmp and cloned_tmp.exists(): shutil.rmtree(cloned_tmp, ignore_errors=True) @@ -125,6 +132,7 @@ def _install_v2_ai( manifest: PackageManifestV2, project_root: Path, target_tools: list[str], + skip_pip: bool = False, ) -> int: """Install using AI-powered adaptation.""" config = load_config() @@ -142,7 +150,7 @@ def _install_v2_ai( _execute_plan(plan, project_root, target_tools) if manifest.mcp_servers: - _install_mcp_servers(manifest, project_root) + _install_mcp_servers(manifest, project_root, skip_pip=skip_pip) console.print(f"\n[green]Installed {manifest.name} successfully.[/green]") return 0 @@ -154,6 +162,7 @@ def _install_v2_fallback( project_root: Path, target_tools: list[str], conflict: str, + skip_pip: bool = False, ) -> int: """Install using file-copy mode (v1 compat or --no-ai).""" installed_count = 0 @@ -205,7 +214,7 @@ def _install_v2_fallback( console.print(f" Installed: {ref.name} → {dest.relative_to(project_root)}") if manifest.mcp_servers: - _install_mcp_servers(manifest, project_root) + _install_mcp_servers(manifest, project_root, skip_pip=skip_pip) console.print(f"\n[green]Installed {installed_count} instructions.[/green]") return 0 @@ -258,17 +267,94 @@ def _get_tool_instruction_path(tool_name: str, project_root: Path, instruction_n return project_root / dir_name / f"{instruction_name}{ext}" -def _install_mcp_servers(manifest: PackageManifestV2, project_root: Path) -> None: - """Install MCP server configurations with credential prompting.""" - servers_with_creds = [s for s in manifest.mcp_servers if s.credentials] +def _install_mcp_servers( + manifest: PackageManifestV2, + project_root: Path, + skip_pip: bool = False, +) -> None: + """Install MCP server configurations with pip dependencies and credential prompting.""" + failed_pip_servers = _install_pip_dependencies(manifest.mcp_servers, skip_pip=skip_pip) + + # Skip credential prompting for servers whose pip deps failed + eligible_servers = [s for s in manifest.mcp_servers if s.name not in failed_pip_servers] + servers_with_creds = [s for s in eligible_servers if s.credentials] if servers_with_creds: env_path = project_root / ".devsync" / ".env" credentials = prompt_mcp_credentials(servers_with_creds, env_path=env_path) - for server in manifest.mcp_servers: + for server in eligible_servers: server_creds = credentials.get(server.name, {}) build_mcp_config(server, server_creds) console.print(f" MCP: {server.name} configured") else: - for server in manifest.mcp_servers: + for server in eligible_servers: console.print(f" MCP: {server.name} (no credentials needed)") + + +def _install_pip_dependencies( + mcp_servers: list[MCPDeclaration], + skip_pip: bool = False, +) -> set[str]: + """Install pip package dependencies for MCP servers. + + Args: + mcp_servers: List of MCPDeclaration objects. + skip_pip: If True, skip all pip installations. + + Returns: + Set of server names whose pip dependency installation failed or was declined. + """ + from devsync.core.pip_utils import ( + get_installed_version, + install_pip_package, + installed_version_satisfies, + validate_pip_spec, + ) + + failed_servers: set[str] = set() + servers_with_pip = [s for s in mcp_servers if s.pip_package] + if not servers_with_pip: + return failed_servers + + console.print("\n[bold]MCP Server Dependencies[/bold]") + + if skip_pip: + console.print(" [yellow]Skipping pip installations (--skip-pip)[/yellow]") + for server in servers_with_pip: + console.print(f" [dim]{server.name}: {server.pip_package} (skipped)[/dim]") + return failed_servers + + for server in servers_with_pip: + spec = server.pip_package + assert spec is not None # guarded by servers_with_pip filter + + if not validate_pip_spec(spec): + console.print(f" [red]Invalid package spec for {server.name}: {spec}[/red]") + failed_servers.add(server.name) + continue + + if installed_version_satisfies(spec): + installed_ver = get_installed_version(spec) + console.print(f" [dim]{server.name}: {spec} already installed (v{installed_ver})[/dim]") + continue + + console.print(f" [cyan]{server.name} requires pip package: {spec}[/cyan]") + if server.description: + console.print(f" [dim]{server.description}[/dim]") + + if not Confirm.ask(f" Install {spec}?", default=True): + console.print(f" [yellow]Skipped pip install for {server.name}[/yellow]") + failed_servers.add(server.name) + continue + + with console.status(f" Installing {spec}..."): + success, message = install_pip_package(spec) + + if success: + console.print(f" [green]{message}[/green]") + else: + console.print(f" [red]{message}[/red]") + console.print(f" [yellow]MCP server {server.name} may not work without {spec}[/yellow]") + failed_servers.add(server.name) + + return failed_servers diff --git a/devsync/cli/main.py b/devsync/cli/main.py index df6749c..ceed233 100644 --- a/devsync/cli/main.py +++ b/devsync/cli/main.py @@ -130,6 +130,11 @@ def install( "-p", help="Target project directory (default: current directory)", ), + skip_pip: bool = typer.Option( + False, + "--skip-pip", + help="Skip pip package installations for MCP servers", + ), ) -> None: """Install a package into the current project. @@ -151,6 +156,9 @@ def install( # Skip conflicts devsync install ./package --conflict skip + + # Skip pip installations + devsync install ./package --skip-pip """ from devsync.cli.install_v2 import install_v2_command @@ -160,6 +168,7 @@ def install( no_ai=no_ai, conflict=conflict, project_dir=project_dir, + skip_pip=skip_pip, ) raise typer.Exit(code=exit_code) diff --git a/devsync/core/component_detector.py b/devsync/core/component_detector.py index a1a3dc7..b92103e 100644 --- a/devsync/core/component_detector.py +++ b/devsync/core/component_detector.py @@ -58,6 +58,7 @@ class DetectedMCPServer: config: dict source: str env_vars: list[str] = field(default_factory=list) + pip_package: Optional[str] = None @dataclass @@ -424,6 +425,10 @@ def _detect_mcp_servers(self) -> list[DetectedMCPServer]: mcp_servers = config_data.get("mcpServers", {}) for server_name, server_config in mcp_servers.items(): env_vars = list(server_config.get("env", {}).keys()) + pip_package = self._resolve_pip_package( + server_config.get("command", ""), + server_config.get("args", []), + ) servers.append( DetectedMCPServer( name=server_name, @@ -431,6 +436,7 @@ def _detect_mcp_servers(self) -> list[DetectedMCPServer]: config=server_config, source=config_location, env_vars=env_vars, + pip_package=pip_package, ) ) except json.JSONDecodeError as e: @@ -445,6 +451,10 @@ def _detect_mcp_servers(self) -> list[DetectedMCPServer]: with open(file_path, "r", encoding="utf-8") as f: server_config = json.load(f) env_vars = list(server_config.get("env", {}).keys()) + pip_package = self._resolve_pip_package( + server_config.get("command", ""), + server_config.get("args", []), + ) servers.append( DetectedMCPServer( name=file_path.stem, @@ -452,6 +462,7 @@ def _detect_mcp_servers(self) -> list[DetectedMCPServer]: config=server_config, source=str(file_path.relative_to(self.project_root)), env_vars=env_vars, + pip_package=pip_package, ) ) except Exception as e: @@ -459,6 +470,24 @@ def _detect_mcp_servers(self) -> list[DetectedMCPServer]: return servers + def _resolve_pip_package(self, command: str, args: list[str]) -> Optional[str]: + """Attempt to resolve a pip package from an MCP server command. + + Non-fatal: returns None on any failure. + + Args: + command: Server executable command. + args: Server command arguments. + + Returns: + Pip package name or None. + """ + if not command: + return None + from devsync.core.pip_utils import resolve_pip_package_for_command + + return resolve_pip_package_for_command(command, args) + def _detect_hooks(self) -> list[DetectedHook]: """Detect hook scripts. diff --git a/devsync/core/extractor.py b/devsync/core/extractor.py index 49c3ca9..ac4c3ac 100644 --- a/devsync/core/extractor.py +++ b/devsync/core/extractor.py @@ -73,7 +73,7 @@ def _read_mcp_configs(self, detection: object) -> list[dict]: configs = [] for server in getattr(detection, "mcp_servers", []): config: dict = {} - for attr in ("name", "command", "args", "env"): + for attr in ("name", "command", "args", "env", "pip_package"): val = getattr(server, attr, None) if val is not None: config[attr] = val @@ -140,6 +140,7 @@ def _extract_without_ai(self, files: dict[str, str], mcp_configs: list[dict]) -> description=f"MCP server: {name}", command=config.get("command", ""), args=config.get("args", []), + pip_package=config.get("pip_package"), ) ) diff --git a/devsync/core/pip_utils.py b/devsync/core/pip_utils.py new file mode 100644 index 0000000..bc7c8a9 --- /dev/null +++ b/devsync/core/pip_utils.py @@ -0,0 +1,314 @@ +"""Pip package validation, detection, and installation utilities. + +All pip-related logic is isolated here for security audit. Functions validate +inputs, detect installed packages, resolve commands to pip packages, and +install packages with comprehensive error handling. +""" + +import importlib.metadata +import logging +import os +import re +import shutil +import subprocess +import sys +from typing import Optional + +logger = logging.getLogger(__name__) + +# Allowlist pattern for pip package specs: name, name>=1.0, name[extra]==2.0, etc. +# Rejects URLs, paths, and shell metacharacters. +_PIP_SPEC_PATTERN = re.compile( + r"^[A-Za-z0-9]([A-Za-z0-9._-]*[A-Za-z0-9])?" # package name + r"(\[[A-Za-z0-9,._-]+\])?" # optional extras + r"([<>=!~]+[A-Za-z0-9.*]+)?" # optional version constraint + r"$" +) + +_DANGEROUS_CHARS = set(";|&$`{}()\n\r") + + +def validate_pip_spec(spec: str) -> bool: + """Validate a pip package specifier against an allowlist. + + Accepts: 'name', 'name>=1.0', 'name[extra]==2.0' + Rejects: URLs, file paths, shell metacharacters, empty strings. + + Args: + spec: Pip package specifier string. + + Returns: + True if the spec is valid and safe. + """ + if not spec or not spec.strip(): + return False + + spec = spec.strip() + + if any(c in spec for c in _DANGEROUS_CHARS): + return False + + if "://" in spec or spec.startswith(("git+", "file:", "/", "\\", ".")): + return False + + return bool(_PIP_SPEC_PATTERN.match(spec)) + + +def _extract_base_name(spec: str) -> str: + """Extract the base package name from a pip spec, stripping version/extras.""" + name = re.split(r"[<>=!~\[]", spec.strip())[0] + return name + + +def is_pip_installed(package_name: str) -> bool: + """Check if a pip package is installed. + + Args: + package_name: Package name (version constraints are stripped). + + Returns: + True if the package is installed. + """ + base = _extract_base_name(package_name) + try: + importlib.metadata.version(base) + return True + except importlib.metadata.PackageNotFoundError: + return False + + +def get_installed_version(package_name: str) -> Optional[str]: + """Get the installed version of a pip package. + + Args: + package_name: Package name (version constraints are stripped). + + Returns: + Version string or None if not installed. + """ + base = _extract_base_name(package_name) + try: + return importlib.metadata.version(base) + except importlib.metadata.PackageNotFoundError: + return None + + +def installed_version_satisfies(spec: str) -> bool: + """Check if the installed version of a package satisfies the spec. + + Args: + spec: Pip package specifier (e.g., 'mcp-server>=1.0'). + + Returns: + True if the package is installed and satisfies the version constraint. + False if not installed or version doesn't satisfy. + """ + base = _extract_base_name(spec) + installed = get_installed_version(base) + if installed is None: + return False + + # Extract version constraint from spec + constraint_match = re.search(r"([<>=!~]+.+)$", spec.strip()) + if not constraint_match: + return True + + constraint = constraint_match.group(1) + + try: + from packaging.specifiers import SpecifierSet + from packaging.version import Version + + specifier = SpecifierSet(constraint) + return Version(installed) in specifier + except ImportError: + # packaging not available — fall back to simple presence check + logger.debug("packaging library not available, skipping version constraint check") + return True + except Exception: + # Invalid version/spec — assume satisfied to avoid blocking install + return True + + +def resolve_pip_package_for_command(command: str, args: list[str]) -> Optional[str]: + """Resolve a command/args pair to a pip package name if possible. + + Patterns detected: + - command="python" (or python3), args contains "-m", "module_name" + - command="uvx", first arg is package name + - command is a console_script entry point + + Args: + command: The executable command. + args: Command arguments. + + Returns: + Pip package name or None if unrecognized. + """ + try: + return _resolve_pip_package_for_command_inner(command, args) + except Exception as e: + logger.warning("Failed to resolve pip package for command %s: %s", command, e) + return None + + +def _resolve_pip_package_for_command_inner(command: str, args: list[str]) -> Optional[str]: + """Inner implementation of resolve_pip_package_for_command.""" + cmd_basename = os.path.basename(command) + + # Pattern 1: python -m module_name + if cmd_basename == "python" or cmd_basename == "python3" or re.match(r"python3\.\d+$", cmd_basename): + if "-m" in args: + m_idx = args.index("-m") + if m_idx + 1 < len(args): + module_name = args[m_idx + 1] + return _find_distribution_for_module(module_name) + + # Pattern 2: uvx package_name + if cmd_basename == "uvx": + if args: + pkg_name = args[0] + if not pkg_name.startswith("-") and validate_pip_spec(pkg_name): + return pkg_name + + # Pattern 3: command is a console_script entry point + dist = _find_distribution_for_script(cmd_basename) + if dist: + return dist + + return None + + +def _find_distribution_for_module(module_name: str) -> Optional[str]: + """Find the distribution that provides a given top-level module. + + Compatible with Python 3.10+ (packages_distributions() is 3.11+). + """ + # Try packages_distributions() (3.11+) + try: + pkg_dists = importlib.metadata.packages_distributions() # type: ignore[attr-defined] + dists = pkg_dists.get(module_name) + if dists: + return dists[0] + except AttributeError: + pass + + # Fallback for 3.10: iterate distributions + for dist in importlib.metadata.distributions(): + top_level = dist.read_text("top_level.txt") + if top_level: + modules = [m.strip() for m in top_level.strip().split("\n") if m.strip()] + if module_name in modules: + return dist.metadata["Name"] + + return None + + +def _find_distribution_for_script(script_name: str) -> Optional[str]: + """Find the distribution that provides a given console_script entry point. + + Compatible with Python 3.10+ (entry_points() API varies by version). + """ + try: + eps = importlib.metadata.entry_points() + # Python 3.12+: eps.select() + if hasattr(eps, "select"): + console_scripts = eps.select(group="console_scripts") # type: ignore[union-attr] + elif isinstance(eps, dict): + # Python 3.10-3.11: eps is a dict + console_scripts = eps.get("console_scripts", []) # type: ignore[arg-type] + else: + console_scripts = [] + + for ep in console_scripts: + if ep.name == script_name: + # ep.dist may not exist on all versions + if hasattr(ep, "dist") and ep.dist is not None: + return ep.dist.metadata["Name"] + continue + except Exception: + pass + + return None + + +def find_pip_executable() -> Optional[str]: + """Find a usable way to run pip. + + Prefers `sys.executable -m pip` (respects current venv), + falls back to `shutil.which("pip")`. + + Returns: + Python executable path (for `-m pip` usage) or standalone pip path, + or None if pip is unavailable. + """ + try: + result = subprocess.run( + [sys.executable, "-m", "pip", "--version"], + capture_output=True, + timeout=10, + check=False, + ) + if result.returncode == 0: + return sys.executable + except (OSError, subprocess.TimeoutExpired): + pass + + pip_path = shutil.which("pip") + if pip_path: + return pip_path + + return None + + +def install_pip_package(spec: str, timeout: int = 120) -> tuple[bool, str]: + """Install a pip package with comprehensive error handling. + + Validates the spec first, then runs pip install in a subprocess. + + Args: + spec: Pip package specifier (e.g., 'mcp-server-github>=1.0'). + timeout: Maximum seconds to wait for install. + + Returns: + Tuple of (success, message). + """ + if not validate_pip_spec(spec): + return (False, f"Invalid pip package spec: {spec}") + + pip_exe = find_pip_executable() + if not pip_exe: + return (False, "pip is not available. Install pip or use a virtual environment.") + + # Build command: either `python -m pip install` or `pip install` + if pip_exe == sys.executable: + cmd = [sys.executable, "-m", "pip", "install", spec] + else: + cmd = [pip_exe, "install", spec] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + + if result.returncode == 0: + return (True, f"Successfully installed {spec}") + + stderr = result.stderr.lower() + if "no matching distribution" in stderr: + return (False, f"Package not found: {spec}") + if "could not find a version" in stderr: + return (False, f"No compatible version found for {spec}") + if "permission denied" in stderr or "permissionerror" in stderr: + return (False, f"Permission denied installing {spec}. Try using a virtual environment.") + + return (False, f"pip install failed for {spec} (exit code {result.returncode})") + + except subprocess.TimeoutExpired: + return (False, f"pip install timed out after {timeout}s for {spec}") + except OSError as e: + return (False, f"Failed to run pip: {e}") diff --git a/devsync/core/practice.py b/devsync/core/practice.py index 765716f..1661117 100644 --- a/devsync/core/practice.py +++ b/devsync/core/practice.py @@ -3,6 +3,8 @@ from dataclasses import dataclass, field from typing import Optional +from devsync.core.pip_utils import validate_pip_spec + @dataclass class CredentialSpec: @@ -135,6 +137,7 @@ class MCPDeclaration: args: list[str] = field(default_factory=list) env_vars: dict[str, str] = field(default_factory=dict) credentials: list[CredentialSpec] = field(default_factory=list) + pip_package: Optional[str] = None def __post_init__(self) -> None: if not self.name: @@ -143,6 +146,9 @@ def __post_init__(self) -> None: raise ValueError("MCPDeclaration description cannot be empty") if self.protocol not in ("stdio", "sse"): raise ValueError(f"MCPDeclaration protocol must be 'stdio' or 'sse', got '{self.protocol}'") + if self.pip_package is not None: + if not validate_pip_spec(self.pip_package): + raise ValueError(f"MCPDeclaration pip_package is not a valid pip spec: '{self.pip_package}'") def to_dict(self) -> dict: result: dict = { @@ -158,6 +164,8 @@ def to_dict(self) -> dict: result["env_vars"] = self.env_vars if self.credentials: result["credentials"] = [c.to_dict() for c in self.credentials] + if self.pip_package is not None: + result["pip_package"] = self.pip_package return result @classmethod @@ -171,4 +179,5 @@ def from_dict(cls, data: dict) -> "MCPDeclaration": args=data.get("args", []), env_vars=data.get("env_vars", {}), credentials=credentials, + pip_package=data.get("pip_package"), ) diff --git a/devsync/llm/prompts.py b/devsync/llm/prompts.py index f262de8..ae0381f 100644 --- a/devsync/llm/prompts.py +++ b/devsync/llm/prompts.py @@ -41,6 +41,9 @@ Input configuration: {mcp_config} +If the MCP server command suggests a pip-installable package (e.g., uvx, python -m), +include the pip_package field with the package name and optional version constraint. + Respond with a JSON object: {{ "name": "server-name", @@ -55,7 +58,8 @@ "description": "what this credential is for", "required": true }} - ] + ], + "pip_package": "package-name>=1.0 (if pip-installable, null otherwise)" }}""" ADAPT_PRACTICE_PROMPT = """\ diff --git a/tests/unit/cli/test_install_v2.py b/tests/unit/cli/test_install_v2.py index 866f4e4..1cc46b8 100644 --- a/tests/unit/cli/test_install_v2.py +++ b/tests/unit/cli/test_install_v2.py @@ -5,7 +5,12 @@ import yaml -from devsync.cli.install_v2 import _get_tool_instruction_path, _resolve_source, install_v2_command +from devsync.cli.install_v2 import ( + _get_tool_instruction_path, + _install_pip_dependencies, + _resolve_source, + install_v2_command, +) class TestResolveSource: @@ -34,6 +39,75 @@ def test_unknown_tool_returns_none(self, tmp_path: Path) -> None: assert result is None +class TestInstallPipDependencies: + def _make_server(self, name: str = "test-mcp", pip_package: str | None = None, description: str = "") -> MagicMock: + server = MagicMock() + server.name = name + server.pip_package = pip_package + server.description = description + return server + + def test_no_pip_servers_returns_empty(self) -> None: + servers = [self._make_server(pip_package=None)] + result = _install_pip_dependencies(servers, skip_pip=False) + assert result == set() + + def test_skip_pip_flag_returns_empty(self) -> None: + servers = [self._make_server(pip_package="mcp-server>=1.0")] + result = _install_pip_dependencies(servers, skip_pip=True) + assert result == set() + + @patch("devsync.core.pip_utils.validate_pip_spec", return_value=False) + def test_invalid_spec_returns_failed(self, mock_validate: MagicMock) -> None: + servers = [self._make_server(name="bad-mcp", pip_package="bad-spec")] + result = _install_pip_dependencies(servers, skip_pip=False) + assert "bad-mcp" in result + + @patch("devsync.core.pip_utils.get_installed_version", return_value="1.2.3") + @patch("devsync.core.pip_utils.installed_version_satisfies", return_value=True) + @patch("devsync.core.pip_utils.validate_pip_spec", return_value=True) + def test_already_installed_skipped( + self, mock_validate: MagicMock, mock_satisfies: MagicMock, mock_version: MagicMock + ) -> None: + servers = [self._make_server(pip_package="mcp-server>=1.0")] + result = _install_pip_dependencies(servers, skip_pip=False) + assert result == set() + + @patch("devsync.cli.install_v2.Confirm.ask", return_value=False) + @patch("devsync.core.pip_utils.installed_version_satisfies", return_value=False) + @patch("devsync.core.pip_utils.validate_pip_spec", return_value=True) + def test_user_declines_returns_failed( + self, mock_validate: MagicMock, mock_satisfies: MagicMock, mock_ask: MagicMock + ) -> None: + servers = [self._make_server(name="declined-mcp", pip_package="mcp-server>=1.0")] + result = _install_pip_dependencies(servers, skip_pip=False) + assert "declined-mcp" in result + + @patch("devsync.core.pip_utils.install_pip_package", return_value=(True, "Successfully installed mcp-server>=1.0")) + @patch("devsync.cli.install_v2.Confirm.ask", return_value=True) + @patch("devsync.core.pip_utils.installed_version_satisfies", return_value=False) + @patch("devsync.core.pip_utils.validate_pip_spec", return_value=True) + def test_install_success( + self, mock_validate: MagicMock, mock_satisfies: MagicMock, mock_ask: MagicMock, mock_install: MagicMock + ) -> None: + servers = [self._make_server(pip_package="mcp-server>=1.0")] + result = _install_pip_dependencies(servers, skip_pip=False) + assert result == set() + mock_install.assert_called_once_with("mcp-server>=1.0") + + @patch("devsync.core.pip_utils.install_pip_package", return_value=(False, "Package not found: bad-pkg")) + @patch("devsync.cli.install_v2.Confirm.ask", return_value=True) + @patch("devsync.core.pip_utils.installed_version_satisfies", return_value=False) + @patch("devsync.core.pip_utils.validate_pip_spec", return_value=True) + def test_install_failure_returns_failed( + self, mock_validate: MagicMock, mock_satisfies: MagicMock, mock_ask: MagicMock, mock_install: MagicMock + ) -> None: + servers = [self._make_server(name="fail-mcp", pip_package="bad-pkg")] + result = _install_pip_dependencies(servers, skip_pip=False) + assert "fail-mcp" in result + mock_install.assert_called_once_with("bad-pkg") + + class TestInstallV2Command: def test_install_nonexistent_source(self) -> None: result = install_v2_command(source="/nonexistent/package/path") diff --git a/tests/unit/core/test_extractor.py b/tests/unit/core/test_extractor.py index 00a909c..a89d51a 100644 --- a/tests/unit/core/test_extractor.py +++ b/tests/unit/core/test_extractor.py @@ -52,6 +52,7 @@ def test_extract_with_mcp_servers(self, tmp_path: Path) -> None: mock_server.command = "npx" mock_server.args = ["-y", "server"] mock_server.env = None + mock_server.pip_package = None with patch("devsync.core.component_detector.ComponentDetector") as mock_cls: mock_cls.return_value.detect_all.return_value = _make_detection_result(mcp_servers=[mock_server]) @@ -63,6 +64,40 @@ def test_extract_with_mcp_servers(self, tmp_path: Path) -> None: assert result.mcp_servers[0].command == "npx" +class TestPracticeExtractorPipPackage: + def test_extract_propagates_pip_package(self, tmp_path: Path) -> None: + mock_server = MagicMock() + mock_server.name = "fetch" + mock_server.command = "python" + mock_server.args = ["-m", "mcp_server_fetch"] + mock_server.env = None + mock_server.pip_package = "mcp-server-fetch>=0.5" + + with patch("devsync.core.component_detector.ComponentDetector") as mock_cls: + mock_cls.return_value.detect_all.return_value = _make_detection_result(mcp_servers=[mock_server]) + extractor = PracticeExtractor(llm_provider=None) + result = extractor.extract(tmp_path) + + assert len(result.mcp_servers) == 1 + assert result.mcp_servers[0].pip_package == "mcp-server-fetch>=0.5" + + def test_extract_pip_package_none_when_absent(self, tmp_path: Path) -> None: + mock_server = MagicMock() + mock_server.name = "github" + mock_server.command = "npx" + mock_server.args = ["-y", "server"] + mock_server.env = None + mock_server.pip_package = None + + with patch("devsync.core.component_detector.ComponentDetector") as mock_cls: + mock_cls.return_value.detect_all.return_value = _make_detection_result(mcp_servers=[mock_server]) + extractor = PracticeExtractor(llm_provider=None) + result = extractor.extract(tmp_path) + + assert len(result.mcp_servers) == 1 + assert result.mcp_servers[0].pip_package is None + + class TestPracticeExtractorWithAI: def test_extract_with_ai(self, tmp_path: Path) -> None: rules_dir = tmp_path / ".claude" / "rules" diff --git a/tests/unit/core/test_pip_utils.py b/tests/unit/core/test_pip_utils.py new file mode 100644 index 0000000..609db73 --- /dev/null +++ b/tests/unit/core/test_pip_utils.py @@ -0,0 +1,288 @@ +"""Tests for pip_utils module.""" + +import importlib.metadata +import subprocess +import sys +from unittest.mock import MagicMock, patch + +from devsync.core.pip_utils import ( + _extract_base_name, + find_pip_executable, + get_installed_version, + install_pip_package, + installed_version_satisfies, + is_pip_installed, + resolve_pip_package_for_command, + validate_pip_spec, +) + + +class TestValidatePipSpec: + def test_valid_simple_name(self) -> None: + assert validate_pip_spec("requests") is True + + def test_valid_hyphenated_name(self) -> None: + assert validate_pip_spec("mcp-server-github") is True + + def test_valid_version_constraint(self) -> None: + assert validate_pip_spec("requests>=2.28") is True + + def test_valid_exact_version(self) -> None: + assert validate_pip_spec("requests==2.28.0") is True + + def test_valid_extras(self) -> None: + assert validate_pip_spec("requests[security]") is True + + def test_valid_extras_with_version(self) -> None: + assert validate_pip_spec("mcp-server[all]>=1.0") is True + + def test_reject_empty(self) -> None: + assert validate_pip_spec("") is False + + def test_reject_whitespace(self) -> None: + assert validate_pip_spec(" ") is False + + def test_reject_git_url(self) -> None: + assert validate_pip_spec("git+https://github.com/user/repo") is False + + def test_reject_file_url(self) -> None: + assert validate_pip_spec("file:///tmp/package.whl") is False + + def test_reject_absolute_path(self) -> None: + assert validate_pip_spec("/tmp/package.whl") is False + + def test_reject_relative_path(self) -> None: + assert validate_pip_spec("./package.whl") is False + + def test_reject_shell_semicolon(self) -> None: + assert validate_pip_spec("requests; rm -rf /") is False + + def test_reject_shell_pipe(self) -> None: + assert validate_pip_spec("requests|evil") is False + + def test_reject_shell_ampersand(self) -> None: + assert validate_pip_spec("requests&evil") is False + + def test_reject_shell_dollar(self) -> None: + assert validate_pip_spec("requests$HOME") is False + + def test_reject_backtick(self) -> None: + assert validate_pip_spec("requests`whoami`") is False + + +class TestExtractBaseName: + def test_simple(self) -> None: + assert _extract_base_name("requests") == "requests" + + def test_with_version(self) -> None: + assert _extract_base_name("requests>=2.28") == "requests" + + def test_with_extras(self) -> None: + assert _extract_base_name("requests[security]") == "requests" + + +class TestIsPipInstalled: + @patch("devsync.core.pip_utils.importlib.metadata.version") + def test_installed(self, mock_version: MagicMock) -> None: + mock_version.return_value = "2.28.0" + assert is_pip_installed("requests") is True + + @patch("devsync.core.pip_utils.importlib.metadata.version") + def test_not_installed(self, mock_version: MagicMock) -> None: + mock_version.side_effect = importlib.metadata.PackageNotFoundError("nope") + assert is_pip_installed("nonexistent-pkg") is False + + @patch("devsync.core.pip_utils.importlib.metadata.version") + def test_strips_version(self, mock_version: MagicMock) -> None: + mock_version.return_value = "1.0" + assert is_pip_installed("requests>=2.0") is True + mock_version.assert_called_once_with("requests") + + +class TestGetInstalledVersion: + @patch("devsync.core.pip_utils.importlib.metadata.version") + def test_found(self, mock_version: MagicMock) -> None: + mock_version.return_value = "1.2.3" + assert get_installed_version("requests") == "1.2.3" + + @patch("devsync.core.pip_utils.importlib.metadata.version") + def test_not_found(self, mock_version: MagicMock) -> None: + mock_version.side_effect = importlib.metadata.PackageNotFoundError("nope") + assert get_installed_version("nonexistent") is None + + +class TestResolvePipPackageForCommand: + @patch("devsync.core.pip_utils._find_distribution_for_module") + def test_python_m_pattern(self, mock_find: MagicMock) -> None: + mock_find.return_value = "mcp-server-fetch" + result = resolve_pip_package_for_command("python", ["-m", "mcp_server_fetch"]) + assert result == "mcp-server-fetch" + mock_find.assert_called_once_with("mcp_server_fetch") + + @patch("devsync.core.pip_utils._find_distribution_for_module") + def test_python3_m_pattern(self, mock_find: MagicMock) -> None: + mock_find.return_value = "some-package" + result = resolve_pip_package_for_command("python3", ["-m", "some_module"]) + assert result == "some-package" + + def test_uvx_pattern(self) -> None: + result = resolve_pip_package_for_command("uvx", ["mcp-server-github"]) + assert result == "mcp-server-github" + + def test_uvx_skips_flags(self) -> None: + result = resolve_pip_package_for_command("uvx", ["--flag", "value"]) + assert result is None + + @patch("devsync.core.pip_utils._find_distribution_for_script") + def test_console_script_pattern(self, mock_find: MagicMock) -> None: + mock_find.return_value = "mcp-server-filesystem" + result = resolve_pip_package_for_command("mcp-server-filesystem", ["--root", "/tmp"]) + assert result == "mcp-server-filesystem" + + @patch("devsync.core.pip_utils._find_distribution_for_script") + def test_unknown_command(self, mock_find: MagicMock) -> None: + mock_find.return_value = None + result = resolve_pip_package_for_command("npx", ["-y", "some-server"]) + assert result is None + + @patch("devsync.core.pip_utils._resolve_pip_package_for_command_inner") + def test_exception_returns_none(self, mock_inner: MagicMock) -> None: + mock_inner.side_effect = RuntimeError("unexpected") + result = resolve_pip_package_for_command("bad", []) + assert result is None + + def test_full_path_python(self) -> None: + with patch("devsync.core.pip_utils._find_distribution_for_module") as mock_find: + mock_find.return_value = "pkg" + result = resolve_pip_package_for_command("/usr/bin/python3", ["-m", "mod"]) + assert result == "pkg" + + +class TestInstalledVersionSatisfies: + @patch("devsync.core.pip_utils.get_installed_version") + def test_not_installed(self, mock_ver: MagicMock) -> None: + mock_ver.return_value = None + assert installed_version_satisfies("requests>=2.0") is False + + @patch("devsync.core.pip_utils.get_installed_version") + def test_no_constraint(self, mock_ver: MagicMock) -> None: + mock_ver.return_value = "1.0.0" + assert installed_version_satisfies("requests") is True + + @patch("devsync.core.pip_utils.get_installed_version") + def test_satisfies_constraint(self, mock_ver: MagicMock) -> None: + mock_ver.return_value = "2.28.0" + assert installed_version_satisfies("requests>=2.0") is True + + @patch("devsync.core.pip_utils.get_installed_version") + def test_does_not_satisfy_constraint(self, mock_ver: MagicMock) -> None: + mock_ver.return_value = "1.5.0" + assert installed_version_satisfies("requests>=2.0") is False + + @patch("devsync.core.pip_utils.get_installed_version") + def test_packaging_not_available(self, mock_ver: MagicMock) -> None: + mock_ver.return_value = "1.0.0" + with patch.dict("sys.modules", {"packaging.specifiers": None, "packaging.version": None}): + # When packaging is unavailable, falls back to True (installed = good enough) + assert installed_version_satisfies("requests>=2.0") is True + + +class TestFindPipExecutable: + @patch("devsync.core.pip_utils.subprocess.run") + def test_sys_executable_works(self, mock_run: MagicMock) -> None: + mock_run.return_value = MagicMock(returncode=0) + result = find_pip_executable() + assert result == sys.executable + + @patch("devsync.core.pip_utils.shutil.which") + @patch("devsync.core.pip_utils.subprocess.run") + def test_fallback_to_which(self, mock_run: MagicMock, mock_which: MagicMock) -> None: + mock_run.return_value = MagicMock(returncode=1) + mock_which.return_value = "/usr/bin/pip" + result = find_pip_executable() + assert result == "/usr/bin/pip" + + @patch("devsync.core.pip_utils.shutil.which") + @patch("devsync.core.pip_utils.subprocess.run") + def test_none_when_unavailable(self, mock_run: MagicMock, mock_which: MagicMock) -> None: + mock_run.side_effect = OSError("no python") + mock_which.return_value = None + result = find_pip_executable() + assert result is None + + +class TestInstallPipPackage: + def test_invalid_spec(self) -> None: + success, msg = install_pip_package("git+https://evil.com/repo") + assert success is False + assert "Invalid pip package spec" in msg + + @patch("devsync.core.pip_utils.find_pip_executable") + def test_no_pip(self, mock_find: MagicMock) -> None: + mock_find.return_value = None + success, msg = install_pip_package("requests") + assert success is False + assert "pip is not available" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_success(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.return_value = MagicMock(returncode=0) + success, msg = install_pip_package("requests>=2.0") + assert success is True + assert "Successfully installed" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_not_found(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.return_value = MagicMock(returncode=1, stderr="no matching distribution found") + success, msg = install_pip_package("nonexistent-pkg") + assert success is False + assert "Package not found" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_permission_denied(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.return_value = MagicMock(returncode=1, stderr="permission denied") + success, msg = install_pip_package("requests") + assert success is False + assert "Permission denied" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_timeout(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.side_effect = subprocess.TimeoutExpired(cmd="pip", timeout=120) + success, msg = install_pip_package("requests", timeout=120) + assert success is False + assert "timed out" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_os_error(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.side_effect = OSError("broken") + success, msg = install_pip_package("requests") + assert success is False + assert "Failed to run pip" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_no_compatible_version(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.return_value = MagicMock(returncode=1, stderr="could not find a version that satisfies") + success, msg = install_pip_package("requests>=999.0") + assert success is False + assert "No compatible version" in msg + + @patch("devsync.core.pip_utils.subprocess.run") + @patch("devsync.core.pip_utils.find_pip_executable") + def test_generic_failure(self, mock_find: MagicMock, mock_run: MagicMock) -> None: + mock_find.return_value = sys.executable + mock_run.return_value = MagicMock(returncode=2, stderr="something unexpected") + success, msg = install_pip_package("requests") + assert success is False + assert "exit code 2" in msg diff --git a/tests/unit/core/test_practice.py b/tests/unit/core/test_practice.py index 43846ab..e4aadc1 100644 --- a/tests/unit/core/test_practice.py +++ b/tests/unit/core/test_practice.py @@ -182,6 +182,57 @@ def test_from_dict(self) -> None: assert len(m.credentials) == 1 assert m.credentials[0].name == "TOKEN" + def test_pip_package_field(self) -> None: + m = MCPDeclaration( + name="fetch", + description="Fetch server", + pip_package="mcp-server-fetch>=0.1", + ) + assert m.pip_package == "mcp-server-fetch>=0.1" + + def test_pip_package_none_by_default(self) -> None: + m = MCPDeclaration(name="test", description="Test") + assert m.pip_package is None + + def test_pip_package_invalid_raises(self) -> None: + with pytest.raises(ValueError, match="not a valid pip spec"): + MCPDeclaration(name="test", description="Test", pip_package="git+https://evil.com") + + def test_pip_package_to_dict_included(self) -> None: + m = MCPDeclaration(name="test", description="Test", pip_package="pkg>=1.0") + d = m.to_dict() + assert d["pip_package"] == "pkg>=1.0" + + def test_pip_package_to_dict_omitted_when_none(self) -> None: + m = MCPDeclaration(name="test", description="Test") + d = m.to_dict() + assert "pip_package" not in d + + def test_pip_package_from_dict(self) -> None: + data = { + "name": "test", + "description": "Test", + "pip_package": "mcp-server>=2.0", + } + m = MCPDeclaration.from_dict(data) + assert m.pip_package == "mcp-server>=2.0" + + def test_pip_package_from_dict_missing(self) -> None: + data = {"name": "test", "description": "Test"} + m = MCPDeclaration.from_dict(data) + assert m.pip_package is None + + def test_pip_package_roundtrip(self) -> None: + original = MCPDeclaration( + name="fetch", + description="Fetch server", + command="python", + args=["-m", "mcp_server_fetch"], + pip_package="mcp-server-fetch>=0.5", + ) + restored = MCPDeclaration.from_dict(original.to_dict()) + assert restored.pip_package == original.pip_package + def test_roundtrip(self) -> None: original = MCPDeclaration( name="db",