diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cc7f021..362e151 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,6 +14,7 @@ jobs: lint-and-test: runs-on: ubuntu-latest strategy: + fail-fast: false matrix: python-version: ["3.10", "3.12"] @@ -40,7 +41,44 @@ jobs: run: ruff check misaka/ - name: MyPy type check + if: matrix.python-version == '3.10' run: mypy misaka/ - name: Run tests run: pytest --cov=misaka --cov-report=term-missing + + environment-check-platforms: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [windows-latest, macos-latest] + python-version: ["3.10", "3.12"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Cache pip packages + uses: actions/cache@v4 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-${{ matrix.python-version }}-${{ hashFiles('pyproject.toml') }} + restore-keys: | + ${{ runner.os }}-pip-${{ matrix.python-version }}- + + - name: Install dependencies + run: pip install -e ".[dev]" + + - name: Run environment setup tests + run: >- + pytest + tests/unit/test_env_check_service.py + tests/unit/test_config_path.py + tests/unit/test_env_check_ui.py + tests/unit/test_update_check_service.py + tests/unit/test_platform.py diff --git a/README.md b/README.md index 61022a4..0001160 100644 --- a/README.md +++ b/README.md @@ -141,8 +141,8 @@ | 依赖 | 要求 | |------|------| | Python | 3.10+ | -| Node.js | 用于 Claude Code CLI | -| Claude Code CLI | `npm install -g @anthropic-ai/claude-code` | +| Node.js | 可选；仅 npm 方式安装 Claude Code 时需要 | +| Claude Code CLI | 推荐使用[官方原生安装方式](https://code.claude.com/docs/en/setup) | | API Key | Anthropic API Key（环境变量或应用内配置） | ### 安装与运行 diff --git a/misaka/config.py b/misaka/config.py index 9b7ed5c..81448a3 100644 --- a/misaka/config.py +++ b/misaka/config.py @@ -119,14 +119,25 @@ def get_extra_path_dirs() -> list[str]: if IS_WINDOWS: appdata = os.environ.get("APPDATA", os.path.join(home, "AppData", "Roaming")) local_appdata = os.environ.get("LOCALAPPDATA", os.path.join(home, "AppData", "Local")) - return [ + program_files = os.environ.get("PROGRAMFILES", r"C:\Program Files") + paths = [ os.path.join(appdata, "npm"), os.path.join(local_appdata, "npm"), + os.path.join(local_appdata, "Microsoft", "WindowsApps"), + os.path.join(local_appdata, "Microsoft", "WinGet", "Links"), + os.path.join(local_appdata, "Programs", "Python"), + os.path.join(program_files, "nodejs"), + os.path.join(program_files, "Git", "cmd"), os.path.join(home, ".npm-global", "bin"), os.path.join(home, ".claude", "bin"), os.path.join(home, ".local", "bin"), os.path.join(home, ".nvm", "current", "bin"), ] + python_root = Path(local_appdata) / "Programs" / "Python" + if python_root.is_dir(): + for python_dir in python_root.glob("Python*"): + paths.extend((str(python_dir), str(python_dir / "Scripts"))) + return paths return [ "/usr/local/bin", "/opt/homebrew/bin", @@ -151,12 +162,42 @@ def get_assets_path() -> Path: def get_expanded_path() -> str: - """Build an expanded PATH that includes common CLI tool locations.""" + """Build a fresh PATH including package-manager changes made after startup.""" current = os.environ.get("PATH", "") parts = [p for p in current.split(os.pathsep) if p] - seen = set(parts) - for p in get_extra_path_dirs(): - if p and p not in seen: - parts.append(p) - seen.add(p) - return os.pathsep.join(parts) + candidates = [*parts, *_get_windows_registry_path_dirs(), *get_extra_path_dirs()] + result: list[str] = [] + seen: set[str] = set() + for path in candidates: + normalized = os.path.expandvars(path.strip().strip('"')) + key = os.path.normcase(normalized) + if normalized and key not in seen: + result.append(normalized) + seen.add(key) + return os.pathsep.join(result) + + +def _get_windows_registry_path_dirs() -> list[str]: + """Read current user/machine PATH values so post-install checks see updates.""" + if not IS_WINDOWS: + return [] + + import winreg + + locations = ( + (winreg.HKEY_CURRENT_USER, r"Environment"), + ( + winreg.HKEY_LOCAL_MACHINE, + r"SYSTEM\CurrentControlSet\Control\Session Manager\Environment", + ), + ) + paths: list[str] = [] + for hive, key_path in locations: + try: + with winreg.OpenKey(hive, key_path) as key: + value, _ = winreg.QueryValueEx(key, "Path") + except OSError: + continue + if isinstance(value, str): + paths.extend(part for part in value.split(os.pathsep) if part) + return paths diff --git a/misaka/i18n/en.json b/misaka/i18n/en.json index 5cc48cc..3c77c77 100644 --- a/misaka/i18n/en.json +++ b/misaka/i18n/en.json @@ -496,7 +496,10 @@ "not_installed": "Not installed", "install": "Install", "installing": "Installing...", + "installing_tool": "Installing {tool}...", + "install_success": "{tool} installed successfully.", "install_failed": "Install failed", + "install_failed_detail": "Installation failed: {error}", "download": "Download", "skip": "Skip", "check_again": "Check Again", diff --git a/misaka/i18n/zh_CN.json b/misaka/i18n/zh_CN.json index 77e40a0..d58fd0b 100644 --- a/misaka/i18n/zh_CN.json +++ b/misaka/i18n/zh_CN.json @@ -496,7 +496,10 @@ "not_installed": "未安装", "install": "安装", "installing": "安装中...", + "installing_tool": "正在安装 {tool}...", + "install_success": "{tool} 安装成功。", "install_failed": "安装失败", + "install_failed_detail": "安装失败：{error}", "download": "下载", "skip": "跳过", "check_again": "重新检查", diff --git a/misaka/i18n/zh_TW.json b/misaka/i18n/zh_TW.json index 15293d0..9c3ed8e 100644 --- a/misaka/i18n/zh_TW.json +++ b/misaka/i18n/zh_TW.json @@ -496,7 +496,10 @@ "not_installed": "未安裝", "install": "安裝", "installing": "安裝中...", + "installing_tool": "正在安裝 {tool}...", + "install_success": "{tool} 安裝成功。", "install_failed": "安裝失敗", + "install_failed_detail": "安裝失敗：{error}", "download": "下載", "skip": "略過", "check_again": "重新檢查", diff --git a/misaka/services/file/update_check_service.py b/misaka/services/file/update_check_service.py index 543a722..ef23002 100644 --- a/misaka/services/file/update_check_service.py +++ b/misaka/services/file/update_check_service.py @@ -8,16 +8,20 @@ from __future__ import annotations import asyncio +import contextlib import json import logging +import os import re +import shutil from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timezone +from pathlib import Path from urllib.error import URLError from urllib.request import Request, urlopen -from misaka.config import get_expanded_path +from misaka.config import IS_MACOS, IS_WINDOWS, get_expanded_path from misaka.utils.platform import ( build_background_subprocess_kwargs, wrap_windows_script_command, @@ -88,40 +92,34 @@ async def perform_update( self, on_progress: Callable[[str], None] | None = None, ) -> bool: - """Update Claude Code CLI to the latest version. - - Runs: npm install -g @anthropic-ai/claude-code@latest - Returns True on success. - After update, clears the cached claude binary path. - """ + """Update Claude Code with its detected installation manager.""" if on_progress: on_progress("Updating Claude Code CLI...") try: - expanded_path = get_expanded_path() - import shutil - - npm_path = shutil.which("npm", path=expanded_path) - if not npm_path: + cmd = self._resolve_update_command() + if not cmd: if on_progress: - on_progress("npm not found, cannot update") + on_progress("Claude Code installation manager was not found") return False - cmd = wrap_windows_script_command( - npm_path, - ["install", "-g", "@anthropic-ai/claude-code@latest"], - ) - proc = await asyncio.create_subprocess_exec( *cmd, + stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + env={**dict(os.environ), "PATH": get_expanded_path()}, **build_background_subprocess_kwargs(), ) - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=300 - ) + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300) + except asyncio.TimeoutError: + with contextlib.suppress(ProcessLookupError): + proc.kill() + with contextlib.suppress(Exception): + await proc.wait() + raise if proc.returncode == 0: # Clear cached claude binary path @@ -156,6 +154,51 @@ async def perform_update( on_progress(f"Update failed: {exc}") return False + def _resolve_update_command(self) -> list[str] | None: + """Select npm, WinGet, Homebrew, or the native updater.""" + from misaka.utils.platform import find_claude_binary + + claude_path = find_claude_binary() + if not claude_path: + return None + + expanded_path = get_expanded_path() + resolved_path = str(Path(claude_path).resolve()).lower() + suffix = Path(claude_path).suffix.lower() + + if "node_modules" in resolved_path or suffix in {".cmd", ".bat", ".ps1"}: + npm_path = shutil.which("npm", path=expanded_path) + if npm_path: + return wrap_windows_script_command( + npm_path, + ["install", "-g", "@anthropic-ai/claude-code@latest"], + ) + + if IS_WINDOWS and ( + "winget" in resolved_path or "windowsapps" in resolved_path + ): + winget_path = shutil.which("winget", path=expanded_path) + if winget_path: + return [ + winget_path, + "upgrade", + "--id", + "Anthropic.ClaudeCode", + "--exact", + "--source", + "winget", + "--accept-source-agreements", + "--accept-package-agreements", + "--disable-interactivity", + ] + + if IS_MACOS and "caskroom" in resolved_path: + brew_path = shutil.which("brew", path=expanded_path) + if brew_path: + return [brew_path, "upgrade", "claude-code"] + + return wrap_windows_script_command(claude_path, ["update"]) + async def _get_current_version(self) -> str | None: """Get the currently installed Claude Code CLI version.""" try: diff --git a/misaka/services/skills/env_check_service.py b/misaka/services/skills/env_check_service.py index ee8e63d..9880abd 100644 --- a/misaka/services/skills/env_check_service.py +++ b/misaka/services/skills/env_check_service.py @@ -1,19 +1,19 @@ -""" -Environment check service for Misaka. - -Detects installed development tools (Claude Code CLI, Node.js, Python, Git) -and provides one-click installation via platform-appropriate methods. -""" +"""Environment detection and guided tool installation for Misaka.""" from __future__ import annotations import asyncio +import contextlib import logging +import os import re +import shlex import shutil +import subprocess from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timezone +from typing import Literal from misaka.config import IS_MACOS, IS_WINDOWS, get_expanded_path from misaka.utils.platform import ( @@ -23,25 +23,31 @@ logger = logging.getLogger(__name__) -# Version extraction pattern +PlatformName = Literal["windows", "macos", "linux"] _VERSION_RE = re.compile(r"v?(\d+\.\d+(?:\.\d+)?)") +_INSTALL_TIMEOUT_SECONDS = 300 +_VERSION_TIMEOUT_SECONDS = 10 + +@dataclass(frozen=True) +class ToolDefinition: + """Definition of a tool checked by :class:`EnvCheckService`.""" -# --------------------------------------------------------------------------- -# Data models -# --------------------------------------------------------------------------- + name: str + commands: tuple[str, ...] + version_flag: str = "--version" @dataclass class ToolStatus: """Status of a single tool dependency.""" - name: str # "Claude Code CLI", "Node.js", "Python", "Git" - command: str # "claude", "node", "python3", "git" - version: str | None # "1.2.3" or None if not found - is_installed: bool # True if binary found and responds to --version - install_url: str # URL for manual download - install_command: str # Platform-specific install command + name: str + command: str + version: str | None + is_installed: bool + install_url: str + install_command: str @dataclass @@ -49,213 +55,234 @@ class EnvironmentCheckResult: """Aggregated result of all environment checks.""" tools: list[ToolStatus] - all_installed: bool # True if every tool is installed - checked_at: str # ISO timestamp - - -# --------------------------------------------------------------------------- -# Tool definitions -# --------------------------------------------------------------------------- - -_TOOL_DEFINITIONS = [ - { - "name": "Claude Code CLI", - "commands": ["claude"], - "version_flag": "--version", - }, - { - "name": "Node.js", - "commands": ["node"], - "version_flag": "--version", - }, - { - "name": "Python", - # On Windows try python first, on Unix try python3 first - "commands": ["python", "python3"] if IS_WINDOWS else ["python3", "python"], - "version_flag": "--version", - }, - { - "name": "Git", - "commands": ["git"], - "version_flag": "--version", - }, -] - - -# --------------------------------------------------------------------------- -# Install info (platform-specific) -# --------------------------------------------------------------------------- + all_installed: bool + checked_at: str + + +@dataclass(frozen=True) +class InstallSpec: + """Executable steps and prerequisites for one platform installer.""" + + steps: tuple[tuple[str, ...], ...] + url: str + required_commands: tuple[str, ...] = () + requires_elevation: bool = False + + +@dataclass(frozen=True) +class InstallResult: + """Structured result returned to UI callers after an install attempt.""" + + tool_name: str + success: bool + message: str + command: str = "" + returncode: int | None = None + + +_TOOL_DEFINITIONS = ( + ToolDefinition("Claude Code CLI", ("claude",)), + ToolDefinition("Node.js", ("node",)), + ToolDefinition( + "Python", + ("py", "python", "python3") if IS_WINDOWS else ("python3", "python"), + ), + ToolDefinition("Git", ("git",)), +) + + +def _current_platform() -> PlatformName: + if IS_WINDOWS: + return "windows" + if IS_MACOS: + return "macos" + return "linux" + + +def _winget_step(package_id: str) -> tuple[str, ...]: + """Build a deterministic, non-interactive WinGet install command.""" + return ( + "winget", + "install", + "--id", + package_id, + "--exact", + "--source", + "winget", + "--accept-source-agreements", + "--accept-package-agreements", + "--disable-interactivity", + ) + + +def _get_install_spec( + tool_name: str, + platform_name: PlatformName | None = None, +) -> InstallSpec | None: + """Return the structured install specification for *tool_name*.""" + platform_name = platform_name or _current_platform() + + urls = { + "Claude Code CLI": "https://code.claude.com/docs/en/setup", + "Node.js": "https://nodejs.org/en/download/", + "Python": "https://www.python.org/downloads/", + "Git": "https://git-scm.com/downloads", + } + url = urls.get(tool_name) + if url is None: + return None + + if platform_name == "windows": + package_ids = { + "Claude Code CLI": "Anthropic.ClaudeCode", + "Node.js": "OpenJS.NodeJS.LTS", + "Python": "Python.Python.3.13", + "Git": "Git.Git", + } + return InstallSpec(steps=(_winget_step(package_ids[tool_name]),), url=url) + + if platform_name == "macos": + brew_args = { + "Claude Code CLI": ("brew", "install", "--cask", "claude-code"), + "Node.js": ("brew", "install", "node"), + "Python": ("brew", "install", "python"), + "Git": ("brew", "install", "git"), + } + return InstallSpec(steps=(brew_args[tool_name],), url=url) -def _get_install_info(tool_name: str) -> tuple[str, str]: - """Return (install_command, install_url) for the given tool. - - Returns platform-specific install commands and download URLs. - """ - - if tool_name == "Node.js": - url = "https://nodejs.org/en/download/" - if IS_WINDOWS: - return ("winget install OpenJS.NodeJS.LTS", url) - elif IS_MACOS: - return ("brew install node", url) - else: - return ("sudo apt install -y nodejs", url) - - if tool_name == "Python": - url = "https://www.python.org/downloads/" - if IS_WINDOWS: - return ("winget install Python.Python.3.12", url) - elif IS_MACOS: - return ("brew install python@3.12", url) - else: - return ("sudo apt install -y python3", url) - - if tool_name == "Git": - url = "https://git-scm.com/downloads" - if IS_WINDOWS: - return ("winget install Git.Git", url) - elif IS_MACOS: - return ("brew install git", url) - else: - return ("sudo apt install -y git", url) if tool_name == "Claude Code CLI": - return ( - "npm install -g @anthropic-ai/claude-code", - "https://docs.anthropic.com/en/docs/claude-code/overview", + return InstallSpec( + steps=(("bash", "-c", "curl -fsSL https://claude.ai/install.sh | bash"),), + url=url, + required_commands=("bash", "curl"), ) - return ("", "") + apt_packages = { + "Node.js": "nodejs", + "Python": "python3", + "Git": "git", + } + package = apt_packages[tool_name] + return InstallSpec( + steps=( + ("apt-get", "update"), + ("apt-get", "install", "-y", package), + ), + url=url, + required_commands=("apt-get",), + requires_elevation=True, + ) + + +def _format_command(command: tuple[str, ...], platform_name: PlatformName) -> str: + if platform_name == "windows": + return subprocess.list2cmdline(command) + return shlex.join(command) -# --------------------------------------------------------------------------- -# Service -# --------------------------------------------------------------------------- +def _get_install_info(tool_name: str) -> tuple[str, str]: + """Return a display command and manual-install URL for *tool_name*.""" + platform_name = _current_platform() + spec = _get_install_spec(tool_name, platform_name) + if spec is None: + return "", "" + command = " && ".join(_format_command(step, platform_name) for step in spec.steps) + return command, spec.url class EnvCheckService: - """Service for checking and installing development tool dependencies.""" + """Check and install the external tools used by Misaka.""" async def check_all(self) -> EnvironmentCheckResult: - """Run all environment checks concurrently. - - Uses asyncio.gather to check all tools in parallel. - Returns an EnvironmentCheckResult with status for each tool. - """ - tasks = [] - for tool_def in _TOOL_DEFINITIONS: - tasks.append( - self._check_tool_multi( - tool_def["name"], - tool_def["commands"], - tool_def["version_flag"], - ) - ) - - results = await asyncio.gather(*tasks, return_exceptions=True) - - tools: list[ToolStatus] = [] - for result in results: - if isinstance(result, Exception): - logger.warning("Tool check failed: %s", result) - tools.append( - ToolStatus( - name="Unknown", - command="", - version=None, - is_installed=False, - install_url="", - install_command="", - ) - ) - else: - tools.append(result) - + """Check all tools concurrently while preserving failure identity.""" + tools = await asyncio.gather( + *(self._check_definition_safe(definition) for definition in _TOOL_DEFINITIONS) + ) return EnvironmentCheckResult( - tools=tools, - all_installed=all(t.is_installed for t in tools), + tools=list(tools), + all_installed=all(tool.is_installed for tool in tools), checked_at=datetime.now(timezone.utc).isoformat(), ) - async def check_tool( - self, command: str, version_flag: str = "--version" - ) -> ToolStatus: - """Check a single tool by running `command version_flag`. - - Handles Windows .cmd/.bat wrappers, expanded PATH lookup. - Parses version string from stdout. - """ - expanded_path = get_expanded_path() - - # Find the binary - binary_path = shutil.which(command, path=expanded_path) - if not binary_path: - install_cmd, install_url = _get_install_info(command) + async def _check_definition_safe(self, definition: ToolDefinition) -> ToolStatus: + try: + return await self._check_tool_multi( + definition.name, + list(definition.commands), + definition.version_flag, + ) + except Exception as exc: + logger.warning("Tool check failed for %s: %s", definition.name, exc) + install_command, install_url = _get_install_info(definition.name) return ToolStatus( - name=command, - command=command, + name=definition.name, + command=definition.commands[0], version=None, is_installed=False, install_url=install_url, - install_command=install_cmd, + install_command=install_command, ) - # Run version check - version = await self._get_version(binary_path, version_flag) - install_cmd, install_url = _get_install_info(command) - return ToolStatus( - name=command, - command=command, - version=version, - is_installed=version is not None, - install_url=install_url, - install_command=install_cmd, + async def check_tool( + self, + command: str, + version_flag: str = "--version", + ) -> ToolStatus: + """Check a single command and return its executable version.""" + definition = next( + (item for item in _TOOL_DEFINITIONS if command in item.commands), + ToolDefinition(command, (command,), version_flag), + ) + return await self._check_tool_multi( + definition.name, + [command], + version_flag, + use_claude_resolver=False, ) async def _check_tool_multi( - self, name: str, commands: list[str], version_flag: str + self, + name: str, + commands: list[str], + version_flag: str, + *, + use_claude_resolver: bool = True, ) -> ToolStatus: - """Check a tool that may have multiple command names. - - Tries each command in order, returns the first one found. - """ + """Try all supported command names and require a parseable version.""" expanded_path = get_expanded_path() - install_cmd, install_url = _get_install_info(name) - - # Special handling for Claude CLI: reuse existing platform utility - if name == "Claude Code CLI": - try: - from misaka.utils.platform import find_claude_binary - - claude_path = find_claude_binary() - if claude_path: - version = await self._get_version(claude_path, version_flag) - if version is None: - version = await self._get_version_lenient(claude_path, version_flag) + install_command, install_url = _get_install_info(name) + + if name == "Claude Code CLI" and use_claude_resolver: + from misaka.utils.platform import find_claude_binary + + claude_path = find_claude_binary() + if claude_path: + version = await self._get_version(claude_path, version_flag) + if version is None: + version = await self._get_version_lenient(claude_path, version_flag) + if version is not None: return ToolStatus( name=name, command="claude", version=version, is_installed=True, install_url=install_url, - install_command=install_cmd, + install_command=install_command, ) - except OSError as exc: - logger.debug("Claude binary check failed: %s", exc) - for cmd in commands: - binary_path = shutil.which(cmd, path=expanded_path) + for command in commands: + binary_path = shutil.which(command, path=expanded_path) if not binary_path: continue - version = await self._get_version(binary_path, version_flag) if version is not None: return ToolStatus( name=name, - command=cmd, + command=command, version=version, is_installed=True, install_url=install_url, - install_command=install_cmd, + install_command=install_command, ) return ToolStatus( @@ -264,168 +291,251 @@ async def _check_tool_multi( version=None, is_installed=False, install_url=install_url, - install_command=install_cmd, + install_command=install_command, ) async def install_tool( self, tool_name: str, on_progress: Callable[[str], None] | None = None, - ) -> bool: - """Install a tool via platform-appropriate method. - - Returns True on success, False on failure. - Calls on_progress with status messages during install. - """ - install_cmd, _ = _get_install_info(tool_name) - if not install_cmd: - logger.warning("No install command for tool: %s", tool_name) - return False + ) -> InstallResult: + """Install a tool and return a structured, user-displayable result.""" + platform_name = _current_platform() + spec = _get_install_spec(tool_name, platform_name) + if spec is None: + return InstallResult(tool_name, False, f"No install command for {tool_name}") + + display_command = " && ".join( + _format_command(step, platform_name) for step in spec.steps + ) + missing = self._find_missing_prerequisite(spec) + if missing: + message = f"Required command not found: {missing}. Manual install: {spec.url}" + self._report_progress(on_progress, message) + return InstallResult(tool_name, False, message, display_command) - if on_progress: - on_progress(f"Installing {tool_name}...") + self._report_progress(on_progress, f"Installing {tool_name}...") + last_returncode: int | None = None try: - args = install_cmd.split() - executable = self._resolve_install_executable(args[0]) - if not executable: - return self._report_install_launcher_missing( - tool_name, args[0], on_progress + for step in spec.steps: + command = self._prepare_install_command(step, spec) + if command is None: + message = ( + "Administrator authorization is unavailable. " + f"Run manually: {display_command}" + ) + self._report_progress(on_progress, message) + return InstallResult(tool_name, False, message, display_command) + + proc = await asyncio.create_subprocess_exec( + *command, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=self._build_install_env(), + **build_background_subprocess_kwargs(), ) + try: + stdout, stderr = await asyncio.wait_for( + proc.communicate(), + timeout=_INSTALL_TIMEOUT_SECONDS, + ) + except asyncio.TimeoutError: + await self._terminate_process(proc) + raise + + last_returncode = proc.returncode + if proc.returncode != 0: + detail = self._decode_process_error(stdout, stderr) + message = f"Install failed: {detail}" + logger.warning( + "Install of %s failed (rc=%s): %s", + tool_name, + proc.returncode, + detail, + ) + self._report_progress(on_progress, message) + return InstallResult( + tool_name, + False, + message, + display_command, + proc.returncode, + ) - cmd = wrap_windows_script_command(executable, args[1:]) - - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - **build_background_subprocess_kwargs(), - ) - - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=300 - ) - - if proc.returncode == 0: - if on_progress: - on_progress(f"{tool_name} installed successfully") - - # Clear cached claude binary path after installation - if tool_name == "Claude Code CLI": - try: - from misaka.utils.platform import clear_claude_cache + if tool_name == "Claude Code CLI": + from misaka.utils.platform import clear_claude_cache - clear_claude_cache() - except (ImportError, OSError) as exc: - logger.debug("Failed to clear claude cache: %s", exc) + clear_claude_cache() - return True - else: - error_msg = stderr.decode(errors="replace").strip() if stderr else "Unknown error" - logger.warning( - "Install of %s failed (rc=%d): %s", + if not await self._verify_installed_tool(tool_name): + message = ( + f"Installer completed, but {tool_name} could not be detected. " + f"Restart Misaka or install manually: {spec.url}" + ) + self._report_progress(on_progress, message) + return InstallResult( tool_name, - proc.returncode, - error_msg, + False, + message, + display_command, + last_returncode, ) - if on_progress: - on_progress(f"Install failed: {error_msg}") - return False + message = f"{tool_name} installed successfully" + self._report_progress(on_progress, message) + return InstallResult( + tool_name, + True, + message, + display_command, + last_returncode, + ) except asyncio.TimeoutError: + message = "Installation timed out" logger.warning("Install of %s timed out", tool_name) - if on_progress: - on_progress("Installation timed out") - return False + self._report_progress(on_progress, message) + return InstallResult(tool_name, False, message, display_command) except Exception as exc: + message = f"Install failed: {exc}" logger.warning("Install of %s failed: %s", tool_name, exc) - if on_progress: - on_progress(f"Install failed: {exc}") - return False - - def _resolve_install_executable(self, executable: str) -> str | None: - """Resolve install launcher from the expanded PATH.""" - expanded_path = get_expanded_path() - return shutil.which(executable, path=expanded_path) - - def _report_install_launcher_missing( + self._report_progress(on_progress, message) + return InstallResult(tool_name, False, message, display_command) + + def _find_missing_prerequisite(self, spec: InstallSpec) -> str | None: + for command in spec.required_commands: + if not self._resolve_install_executable(command): + return command + for step in spec.steps: + if not self._resolve_install_executable(step[0]): + return step[0] + return None + + def _prepare_install_command( self, - tool_name: str, - executable: str, - on_progress: Callable[[str], None] | None, - ) -> bool: - """Report a missing launcher with a user-friendly message.""" - logger.warning( - "Install of %s cannot start because launcher was not found: %s", - tool_name, - executable, - ) - if on_progress: - on_progress(f"Install failed: required command not found: {executable}") - return False + step: tuple[str, ...], + spec: InstallSpec, + ) -> list[str] | None: + executable = self._resolve_install_executable(step[0]) + if executable is None: + return None + command = [executable, *step[1:]] - async def _get_version( - self, binary_path: str, version_flag: str - ) -> str | None: - """Run a binary with its version flag and parse the version string.""" - try: - cmd = wrap_windows_script_command(binary_path, [version_flag]) + if spec.requires_elevation and not self._is_root(): + elevation = self._resolve_elevation_command() + if elevation is None: + return None + command = [*elevation, *command] - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - **build_background_subprocess_kwargs(), - ) + return wrap_windows_script_command(command[0], command[1:]) - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=10 - ) + @staticmethod + def _is_root() -> bool: + get_euid = getattr(os, "geteuid", None) + return bool(get_euid and get_euid() == 0) - if proc.returncode != 0: - return None + def _resolve_elevation_command(self) -> list[str] | None: + expanded_path = get_expanded_path() + if os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"): + pkexec = shutil.which("pkexec", path=expanded_path) + if pkexec: + return [pkexec] + sudo = shutil.which("sudo", path=expanded_path) + if sudo: + return [sudo, "--non-interactive"] + return None + + @staticmethod + def _build_install_env() -> dict[str, str]: + env = os.environ.copy() + env["PATH"] = get_expanded_path() + env.setdefault("HOMEBREW_NO_AUTO_UPDATE", "1") + env.setdefault("DEBIAN_FRONTEND", "noninteractive") + return env + + @staticmethod + async def _terminate_process(proc: asyncio.subprocess.Process) -> None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + with contextlib.suppress(Exception): + await proc.wait() + + @staticmethod + def _decode_process_error(stdout: bytes, stderr: bytes) -> str: + output = (stderr or stdout).decode(errors="replace").strip() + if not output: + return "Unknown error" + return output[-2000:] + + @staticmethod + def _report_progress( + callback: Callable[[str], None] | None, + message: str, + ) -> None: + if not callback: + return + try: + callback(message) + except Exception: + logger.debug("Install progress callback failed", exc_info=True) - output = (stdout or b"").decode(errors="replace") - # Some tools output version to stderr (e.g., python --version on some systems) - if not output.strip(): - output = (stderr or b"").decode(errors="replace") + def _resolve_install_executable(self, executable: str) -> str | None: + return shutil.which(executable, path=get_expanded_path()) - match = _VERSION_RE.search(output) - return match.group(1) if match else None + async def _verify_installed_tool(self, tool_name: str) -> bool: + definition = next( + (item for item in _TOOL_DEFINITIONS if item.name == tool_name), + None, + ) + if definition is None: + return False + status = await self._check_definition_safe(definition) + return status.is_installed - except (asyncio.TimeoutError, OSError, Exception) as exc: - logger.debug("Version check failed for %s: %s", binary_path, exc) - return None + async def _get_version(self, binary_path: str, version_flag: str) -> str | None: + return await self._capture_version(binary_path, version_flag, require_success=True) async def _get_version_lenient( - self, binary_path: str, version_flag: str + self, + binary_path: str, + version_flag: str, ) -> str | None: - """Like _get_version but ignores non-zero exit codes. + return await self._capture_version(binary_path, version_flag, require_success=False) - Some CLI wrappers (e.g. .cmd on Windows) may report a non-zero - return code even though they print valid version output. - """ + async def _capture_version( + self, + binary_path: str, + version_flag: str, + *, + require_success: bool, + ) -> str | None: + proc: asyncio.subprocess.Process | None = None try: - cmd = wrap_windows_script_command(binary_path, [version_flag]) - + command = wrap_windows_script_command(binary_path, [version_flag]) proc = await asyncio.create_subprocess_exec( - *cmd, + *command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + env=self._build_install_env(), **build_background_subprocess_kwargs(), ) - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=10 + proc.communicate(), + timeout=_VERSION_TIMEOUT_SECONDS, + ) + if require_success and proc.returncode != 0: + return None + output = "\n".join( + part.decode(errors="replace") for part in (stdout, stderr) if part ) - - output = (stdout or b"").decode(errors="replace") - if not output.strip(): - output = (stderr or b"").decode(errors="replace") - match = _VERSION_RE.search(output) return match.group(1) if match else None - - except (asyncio.TimeoutError, OSError, Exception) as exc: - logger.debug("Lenient version check failed for %s: %s", binary_path, exc) + except asyncio.TimeoutError: + if proc is not None: + await self._terminate_process(proc) + logger.debug("Version check timed out for %s", binary_path) + return None + except Exception as exc: + logger.debug("Version check failed for %s: %s", binary_path, exc) return None diff --git a/misaka/ui/common/app_shell.py b/misaka/ui/common/app_shell.py index 88c7afb..6b83ea3 100644 --- a/misaka/ui/common/app_shell.py +++ b/misaka/ui/common/app_shell.py @@ -14,6 +14,7 @@ import flet as ft +from misaka.i18n import t from misaka.ui.chat.pages.chat_page import ChatPage from misaka.ui.common.theme import apply_theme from misaka.ui.dashboard.pages.dashboard_page import DashboardPage @@ -276,17 +277,28 @@ def show_env_check_dialog(self) -> None: def _handle_env_install(self, tool_name: str) -> None: """Handle tool install request from env check dialog.""" - env_svc = self.state.get_service('env_check_service') - if env_svc: - async def _do_install(): - await env_svc.install_tool(tool_name) - result = await env_svc.check_all() - self.state.env_check_result = result - if self._env_check_dialog: - self._env_check_dialog.refresh() - self.state.update() + env_svc = self.state.get_service("env_check_service") + if not env_svc or not self.state.page: + return - self.state.page.run_task(_do_install) + def _on_progress(message: str) -> None: + if self._env_check_dialog: + self._env_check_dialog.set_install_progress( + t("env_check.installing_tool", tool=tool_name) + ) + self.state.update() + + async def _do_install() -> None: + install_result = await env_svc.install_tool( + tool_name, + on_progress=_on_progress, + ) + self.state.env_check_result = await env_svc.check_all() + if self._env_check_dialog: + self._env_check_dialog.finish_install(install_result) + self.state.update() + + self.state.page.run_task(_do_install) def _dismiss_env_check(self) -> None: """Dismiss the environment check dialog.""" @@ -299,7 +311,7 @@ def _dismiss_env_check(self) -> None: def _recheck_env(self) -> None: """Re-run environment checks.""" - env_svc = self.state.get_service('env_check_service') + env_svc = self.state.get_service("env_check_service") if env_svc: async def _do_recheck(): await asyncio.sleep(0) diff --git a/misaka/ui/dialogs/env_check_dialog.py b/misaka/ui/dialogs/env_check_dialog.py index 8e55858..9e3905d 100644 --- a/misaka/ui/dialogs/env_check_dialog.py +++ b/misaka/ui/dialogs/env_check_dialog.py @@ -7,6 +7,7 @@ from __future__ import annotations +import contextlib from collections.abc import Callable from typing import TYPE_CHECKING @@ -25,6 +26,7 @@ ) if TYPE_CHECKING: + from misaka.services.skills.env_check_service import InstallResult from misaka.state import AppState @@ -52,6 +54,8 @@ def __init__( self._on_dismiss = on_dismiss self._on_recheck = on_recheck self._installing_tool: str | None = None + self._status_message: str | None = None + self._status_is_error = False self._build_ui() def _build_ui(self) -> None: @@ -85,6 +89,32 @@ def _build_ui(self) -> None: visible=check_result.all_installed, ) + install_status = ft.Container( + content=ft.Row( + controls=[ + ft.Icon( + ft.Icons.ERROR if self._status_is_error else ft.Icons.INFO, + color=ERROR_RED if self._status_is_error else ft.Colors.PRIMARY, + size=18, + ), + ft.Text( + self._status_message or "", + size=12, + color=ERROR_RED if self._status_is_error else None, + expand=True, + ), + ], + spacing=8, + ), + padding=ft.Padding.symmetric(horizontal=12, vertical=8), + border_radius=RADIUS_LG, + bgcolor=ft.Colors.with_opacity( + 0.08, + ERROR_RED if self._status_is_error else ft.Colors.PRIMARY, + ), + visible=bool(self._status_message), + ) + skip_btn = make_outlined_button( t("env_check.skip"), on_click=self._handle_dismiss, @@ -137,6 +167,7 @@ def _build_ui(self) -> None: controls=tool_cards, spacing=8, ), + install_status, all_ready_msg, make_divider(), actions, @@ -210,6 +241,7 @@ def _build_tool_card(self, tool) -> ft.Control: icon=ft.Icons.DOWNLOAD, on_click=lambda e, name=tool.name: self._handle_install(name), ) + status_badge.disabled = self._installing_tool is not None return ft.Container( content=ft.Row( @@ -246,8 +278,14 @@ def _build_tool_card(self, tool) -> ft.Control: ) def _handle_install(self, tool_name: str) -> None: + if self._installing_tool is not None: + return self._installing_tool = tool_name + self._status_message = None + self._status_is_error = False self._build_ui() + with contextlib.suppress(AssertionError, RuntimeError): + self.update() if self._on_install: self._on_install(tool_name) @@ -259,7 +297,26 @@ def _handle_recheck(self, e: ft.ControlEvent | None = None) -> None: if self._on_recheck: self._on_recheck() + def set_install_progress(self, message: str) -> None: + """Display an install progress message without rebuilding the page.""" + self._status_message = message + self._status_is_error = False + self._build_ui() + + def finish_install(self, result: InstallResult) -> None: + """Finish the active install and retain success/failure feedback.""" + self._installing_tool = None + self._status_message = ( + t("env_check.install_success", tool=result.tool_name) + if result.success + else t("env_check.install_failed_detail", error=result.message) + ) + self._status_is_error = not result.success + self._build_ui() + def refresh(self, check_result=None) -> None: """Update the dialog after an install attempt or recheck.""" self._installing_tool = None + self._status_message = None + self._status_is_error = False self._build_ui() diff --git a/misaka/ui/settings/components/env_status_panel.py b/misaka/ui/settings/components/env_status_panel.py index 44a2bd7..9356d3d 100644 --- a/misaka/ui/settings/components/env_status_panel.py +++ b/misaka/ui/settings/components/env_status_panel.py @@ -18,6 +18,7 @@ if TYPE_CHECKING: from misaka.db.database import DatabaseBackend + from misaka.services.skills.env_check_service import InstallResult from misaka.state import AppState @@ -35,6 +36,8 @@ def __init__( self.padding = ft.Padding.symmetric(horizontal=24, vertical=16) self._env_checking: bool = False self._env_installing_tool: str | None = None + self._install_message: str | None = None + self._install_error = False self._build_ui() def refresh(self) -> None: @@ -65,12 +68,40 @@ def _build_ui(self) -> None: ), ft.Text(t("settings.env_status_desc"), size=12, opacity=0.6), ft.Column(controls=tool_rows, spacing=8), + self._build_install_status(), ], spacing=12, scroll=ft.ScrollMode.AUTO, expand=True, ) + def _build_install_status(self) -> ft.Control: + return ft.Container( + content=ft.Row( + controls=[ + ft.Icon( + ft.Icons.ERROR if self._install_error else ft.Icons.INFO, + color=ERROR_RED if self._install_error else ft.Colors.PRIMARY, + size=18, + ), + ft.Text( + self._install_message or "", + color=ERROR_RED if self._install_error else None, + size=12, + expand=True, + ), + ], + spacing=8, + ), + padding=ft.Padding.symmetric(horizontal=12, vertical=8), + border_radius=RADIUS_LG, + bgcolor=ft.Colors.with_opacity( + 0.08, + ERROR_RED if self._install_error else ft.Colors.PRIMARY, + ), + visible=bool(self._install_message), + ) + def _build_header_button(self) -> ft.Control: if self._env_checking: return ft.Row( @@ -150,11 +181,13 @@ def _build_action_widget(self, tool, is_installed: bool, is_installing: bool) -> ], spacing=6, ) - return make_button( + button = make_button( t("env_check.install"), icon=ft.Icons.DOWNLOAD, on_click=lambda e, name=tool.name: self._handle_install(e, name), ) + button.disabled = self._env_installing_tool is not None + return button # ------------------------------------------------------------------ # Event handlers @@ -165,6 +198,8 @@ def _handle_recheck(self, e: ft.ControlEvent) -> None: if not page: return self._env_checking = True + self._install_message = None + self._install_error = False self._build_ui() self.state.update() page.run_task(self._do_recheck) @@ -182,6 +217,8 @@ def _handle_install(self, e: ft.ControlEvent, tool_name: str) -> None: if not page: return self._env_installing_tool = tool_name + self._install_message = None + self._install_error = False self._build_ui() self.state.update() @@ -193,8 +230,26 @@ async def _install_task() -> None: async def _do_install(self, tool_name: str) -> None: svc = self.state.get_service("env_check_service") if svc: - await svc.install_tool(tool_name) + result = await svc.install_tool(tool_name, on_progress=self._on_install_progress) self.state.env_check_result = await svc.check_all() + self._finish_install(result) self._env_installing_tool = None self._build_ui() self.state.update() + + def _on_install_progress(self, message: str) -> None: + self._install_message = t( + "env_check.installing_tool", + tool=self._env_installing_tool or "", + ) + self._install_error = False + self._build_ui() + self.state.update() + + def _finish_install(self, result: InstallResult) -> None: + self._install_message = ( + t("env_check.install_success", tool=result.tool_name) + if result.success + else t("env_check.install_failed_detail", error=result.message) + ) + self._install_error = not result.success diff --git a/misaka/utils/platform.py b/misaka/utils/platform.py index 020b063..f100254 100644 --- a/misaka/utils/platform.py +++ b/misaka/utils/platform.py @@ -10,6 +10,7 @@ import asyncio import logging import os +import re import shutil import subprocess from pathlib import Path @@ -17,6 +18,7 @@ from misaka.config import IS_WINDOWS, get_expanded_path logger = logging.getLogger(__name__) +_CLAUDE_VERSION_RE = re.compile(r"v?\d+\.\d+(?:\.\d+)?") # --------------------------------------------------------------------------- @@ -213,13 +215,16 @@ def _validate_claude_binary(path: str) -> bool: return False try: command = wrap_windows_script_command(path, ["--version"]) - subprocess.run( + result = subprocess.run( command, capture_output=True, timeout=5, **build_background_subprocess_kwargs(), ) - return True + output = b"\n".join((result.stdout or b"", result.stderr or b"")).decode( + errors="replace" + ) + return bool(_CLAUDE_VERSION_RE.search(output)) except (subprocess.SubprocessError, OSError): return False diff --git a/tests/unit/test_config_path.py b/tests/unit/test_config_path.py new file mode 100644 index 0000000..e1a9880 --- /dev/null +++ b/tests/unit/test_config_path.py @@ -0,0 +1,65 @@ +"""Tests for runtime PATH refresh after package-manager installs.""" + +from __future__ import annotations + +import os +import sys +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from misaka import config + + +def test_expanded_path_merges_fresh_registry_entries() -> None: + with patch.dict(os.environ, {"PATH": os.pathsep.join(("old-bin", "shared-bin"))}), patch( + "misaka.config._get_windows_registry_path_dirs", + return_value=["new-bin", "shared-bin"], + ), patch("misaka.config.get_extra_path_dirs", return_value=["extra-bin"]): + result = config.get_expanded_path().split(os.pathsep) + + assert result == ["old-bin", "shared-bin", "new-bin", "extra-bin"] + + +def test_windows_registry_path_reader_uses_user_and_machine_values() -> None: + key = MagicMock() + key.__enter__.return_value = key + fake_winreg = SimpleNamespace( + HKEY_CURRENT_USER=object(), + HKEY_LOCAL_MACHINE=object(), + OpenKey=MagicMock(return_value=key), + QueryValueEx=MagicMock( + side_effect=[ + (os.pathsep.join(("user-a", "user-b")), 1), + ("machine-a", 1), + ] + ), + ) + + with patch.object(config, "IS_WINDOWS", True), patch.dict( + sys.modules, + {"winreg": fake_winreg}, + ): + result = config._get_windows_registry_path_dirs() + + assert result == ["user-a", "user-b", "machine-a"] + assert fake_winreg.OpenKey.call_count == 2 + + +def test_windows_extra_paths_include_package_manager_locations() -> None: + environment = { + "APPDATA": r"C:\Users\tester\AppData\Roaming", + "LOCALAPPDATA": r"C:\Users\tester\AppData\Local", + "PROGRAMFILES": r"C:\Program Files", + } + with patch.object(config, "IS_WINDOWS", True), patch.dict( + os.environ, + environment, + clear=False, + ), patch("pathlib.Path.is_dir", return_value=False): + paths = config.get_extra_path_dirs() + + normalized = [path.replace("\\", "/") for path in paths] + assert "C:/Program Files/nodejs" in normalized + assert "C:/Program Files/Git/cmd" in normalized + assert any(path.endswith("Microsoft/WindowsApps") for path in normalized) + assert any(path.endswith("Microsoft/WinGet/Links") for path in normalized) diff --git a/tests/unit/test_env_check_service.py b/tests/unit/test_env_check_service.py index ae4d71c..f4500bc 100644 --- a/tests/unit/test_env_check_service.py +++ b/tests/unit/test_env_check_service.py @@ -1,20 +1,19 @@ -""" -Tests for the EnvCheckService. -""" +"""Tests for cross-platform environment detection and installation.""" from __future__ import annotations import asyncio -import subprocess -from unittest.mock import ANY, AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from misaka.services.skills.env_check_service import ( EnvCheckService, EnvironmentCheckResult, + InstallResult, ToolStatus, _get_install_info, + _get_install_spec, ) @@ -23,495 +22,469 @@ def service() -> EnvCheckService: return EnvCheckService() -class TestToolStatus: - - def test_tool_status_dataclass(self) -> None: - status = ToolStatus( - name="Node.js", - command="node", - version="20.11.1", - is_installed=True, - install_url="https://nodejs.org", - install_command="brew install node", - ) - assert status.name == "Node.js" +def _process( + *, + stdout: bytes = b"", + stderr: bytes = b"", + returncode: int = 0, + communicate_error: Exception | None = None, +) -> MagicMock: + proc = MagicMock() + if communicate_error is None: + proc.communicate = AsyncMock(return_value=(stdout, stderr)) + else: + proc.communicate = AsyncMock(side_effect=communicate_error) + proc.wait = AsyncMock(return_value=returncode) + proc.kill = MagicMock() + proc.returncode = returncode + return proc + + +class TestDataModels: + def test_tool_status(self) -> None: + status = ToolStatus("Node.js", "node", "24.1.0", True, "url", "command") assert status.is_installed is True - assert status.version == "20.11.1" - - -class TestEnvironmentCheckResult: + assert status.version == "24.1.0" - def test_all_installed_true(self) -> None: - tools = [ - ToolStatus("A", "a", "1.0", True, "", ""), - ToolStatus("B", "b", "2.0", True, "", ""), - ] - result = EnvironmentCheckResult( - tools=tools, all_installed=True, checked_at="2026-01-01T00:00:00" - ) + def test_environment_result(self) -> None: + result = EnvironmentCheckResult([], True, "2026-01-01T00:00:00Z") assert result.all_installed is True - def test_all_installed_false(self) -> None: - tools = [ - ToolStatus("A", "a", "1.0", True, "", ""), - ToolStatus("B", "b", None, False, "", ""), - ] - result = EnvironmentCheckResult( - tools=tools, all_installed=False, checked_at="2026-01-01T00:00:00" + def test_install_result(self) -> None: + result = InstallResult("Git", False, "failed", "git install", 1) + assert result.success is False + assert result.returncode == 1 + + +class TestInstallSpecs: + @pytest.mark.parametrize( + ("tool_name", "package_id"), + [ + ("Claude Code CLI", "Anthropic.ClaudeCode"), + ("Node.js", "OpenJS.NodeJS.LTS"), + ("Python", "Python.Python.3.13"), + ("Git", "Git.Git"), + ], + ) + def test_windows_specs_are_exact_and_non_interactive( + self, + tool_name: str, + package_id: str, + ) -> None: + spec = _get_install_spec(tool_name, "windows") + assert spec is not None + step = spec.steps[0] + assert step[:4] == ("winget", "install", "--id", package_id) + assert "--exact" in step + assert "--source" in step + assert "--accept-source-agreements" in step + assert "--accept-package-agreements" in step + assert "--disable-interactivity" in step + + @pytest.mark.parametrize( + ("tool_name", "expected"), + [ + ("Claude Code CLI", ("brew", "install", "--cask", "claude-code")), + ("Node.js", ("brew", "install", "node")), + ("Python", ("brew", "install", "python")), + ("Git", ("brew", "install", "git")), + ], + ) + def test_macos_specs_use_homebrew( + self, + tool_name: str, + expected: tuple[str, ...], + ) -> None: + spec = _get_install_spec(tool_name, "macos") + assert spec is not None + assert spec.steps == (expected,) + + @pytest.mark.parametrize( + ("tool_name", "package"), + [("Node.js", "nodejs"), ("Python", "python3"), ("Git", "git")], + ) + def test_linux_specs_use_apt_get(self, tool_name: str, package: str) -> None: + spec = _get_install_spec(tool_name, "linux") + assert spec is not None + assert spec.steps == ( + ("apt-get", "update"), + ("apt-get", "install", "-y", package), ) - assert result.all_installed is False - - -class TestGetInstallInfo: - - def test_claude_cli_install_info(self) -> None: - cmd, url = _get_install_info("Claude Code CLI") - assert "npm install -g" in cmd - assert "claude-code" in cmd - assert url != "" - - def test_nodejs_install_info(self) -> None: - cmd, url = _get_install_info("Node.js") - assert cmd != "" - assert "nodejs.org" in url - - def test_python_install_info(self) -> None: - cmd, url = _get_install_info("Python") - assert cmd != "" - assert "python.org" in url + assert spec.requires_elevation is True - def test_git_install_info(self) -> None: - cmd, url = _get_install_info("Git") - assert cmd != "" - assert "git-scm.com" in url + def test_linux_claude_uses_official_native_installer(self) -> None: + spec = _get_install_spec("Claude Code CLI", "linux") + assert spec is not None + assert "https://claude.ai/install.sh" in spec.steps[0][-1] + assert spec.required_commands == ("bash", "curl") - def test_unknown_tool_returns_empty(self) -> None: - cmd, url = _get_install_info("UnknownTool") - assert cmd == "" - assert url == "" + def test_unknown_tool_has_no_spec(self) -> None: + assert _get_install_spec("Unknown", "windows") is None + assert _get_install_info("Unknown") == ("", "") -class TestEnvCheckService: - +class TestDetection: async def test_check_tool_found(self, service: EnvCheckService) -> None: - """check_tool returns is_installed=True when binary exists and responds.""" - with patch("shutil.which", return_value="/usr/bin/node"), \ - patch.object(service, "_get_version", return_value="20.11.1"): - result = await service.check_tool("node", "--version") - assert result.is_installed is True - assert result.version == "20.11.1" + with patch( + "misaka.services.skills.env_check_service.shutil.which", + return_value="/usr/bin/node", + ), patch.object(service, "_get_version", return_value="24.1.0"): + result = await service.check_tool("node") + assert result.name == "Node.js" + assert result.is_installed is True + assert result.version == "24.1.0" async def test_check_tool_not_found(self, service: EnvCheckService) -> None: - """check_tool returns is_installed=False when binary is missing.""" - with patch("shutil.which", return_value=None): - result = await service.check_tool("nonexistent") - assert result.is_installed is False - assert result.version is None - - async def test_check_tool_found_but_version_fails(self, service: EnvCheckService) -> None: - """check_tool returns is_installed=False when version check fails.""" - with patch("shutil.which", return_value="/usr/bin/node"), \ - patch.object(service, "_get_version", return_value=None): - result = await service.check_tool("node", "--version") - assert result.is_installed is False - - async def test_check_all_returns_four_tools(self, service: EnvCheckService) -> None: - """check_all should return results for all 4 tools.""" - mock_status = ToolStatus( - name="test", command="test", version="1.0", - is_installed=True, install_url="", install_command="", - ) - with patch.object(service, "_check_tool_multi", return_value=mock_status): - result = await service.check_all() - assert len(result.tools) == 4 - assert result.all_installed is True - assert result.checked_at != "" - - async def test_check_all_with_missing_tool(self, service: EnvCheckService) -> None: - """check_all sets all_installed=False when any tool is missing.""" - installed = ToolStatus( - name="test", command="test", version="1.0", - is_installed=True, install_url="", install_command="", - ) - not_installed = ToolStatus( - name="missing", command="missing", version=None, - is_installed=False, install_url="", install_command="", - ) - - call_count = 0 - - async def mock_check(*args, **kwargs): - nonlocal call_count - call_count += 1 - return not_installed if call_count == 2 else installed - - with patch.object(service, "_check_tool_multi", side_effect=mock_check): - result = await service.check_all() - assert result.all_installed is False - - async def test_check_all_handles_exception(self, service: EnvCheckService) -> None: - """check_all should handle exceptions from individual tool checks.""" - async def mock_check(*args, **kwargs): - raise RuntimeError("test error") - - with patch.object(service, "_check_tool_multi", side_effect=mock_check): - result = await service.check_all() - assert len(result.tools) == 4 - assert result.all_installed is False - - async def test_get_version_parses_standard_output(self, service: EnvCheckService) -> None: - """_get_version should parse version from standard output.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"v20.11.1\n", b"") - mock_proc.returncode = 0 - - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version("/usr/bin/node", "--version") - assert version == "20.11.1" - - async def test_get_version_parses_git_output(self, service: EnvCheckService) -> None: - """_get_version should parse version from git-style output.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"git version 2.43.0\n", b"") - mock_proc.returncode = 0 - - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version("/usr/bin/git", "--version") - assert version == "2.43.0" - - async def test_get_version_returns_none_on_failure(self, service: EnvCheckService) -> None: - """_get_version should return None when the command fails.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"", b"error") - mock_proc.returncode = 1 - - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version("/usr/bin/nonexistent", "--version") - assert version is None - - async def test_get_version_returns_none_on_timeout(self, service: EnvCheckService) -> None: - """_get_version should return None on timeout.""" - with patch("asyncio.create_subprocess_exec", side_effect=asyncio.TimeoutError()): - version = await service._get_version("/usr/bin/slow", "--version") - assert version is None - - async def test_get_version_reads_stderr_fallback(self, service: EnvCheckService) -> None: - """_get_version should read stderr when stdout is empty.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"", b"Python 3.12.1\n") - mock_proc.returncode = 0 - - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version("/usr/bin/python3", "--version") - assert version == "3.12.1" - - async def test_install_tool_success(self, service: EnvCheckService) -> None: - """install_tool should use shared hidden subprocess helpers.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"installed\n", b"") - mock_proc.returncode = 0 - - progress_messages: list[str] = [] - with patch( - "misaka.services.skills.env_check_service.IS_WINDOWS", True - ), patch( - "shutil.which", - return_value="C:\\Windows\\System32\\winget.exe", - ) as mock_which, patch( - "misaka.services.skills.env_check_service.wrap_windows_script_command", - return_value=[ - "C:\\Windows\\System32\\winget.exe", - "install", - "OpenJS.NodeJS.LTS", - ], - ) as mock_wrap, patch( - "misaka.services.skills.env_check_service.build_background_subprocess_kwargs", - return_value={"creationflags": 1, "startupinfo": "hidden"}, - ) as mock_kwargs, patch( - "asyncio.create_subprocess_exec", return_value=mock_proc - ) as mock_exec: - result = await service.install_tool( - "Node.js", on_progress=progress_messages.append - ) - assert result is True - assert any("successfully" in m for m in progress_messages) - mock_which.assert_called_once() - mock_wrap.assert_called_once_with( - "C:\\Windows\\System32\\winget.exe", - ["install", "OpenJS.NodeJS.LTS"], - ) - mock_kwargs.assert_called_once_with() - mock_exec.assert_called_once_with( - "C:\\Windows\\System32\\winget.exe", - "install", - "OpenJS.NodeJS.LTS", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - creationflags=1, - startupinfo="hidden", - ) + "misaka.services.skills.env_check_service.shutil.which", + return_value=None, + ): + result = await service.check_tool("node") + assert result.is_installed is False - async def test_install_tool_resolves_npm_cmd_for_claude_cli( - self, service: EnvCheckService - ) -> None: - """install_tool should resolve npm to npm.cmd before launching.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"installed\n", b"") - mock_proc.returncode = 0 + async def test_python_falls_back_to_second_command(self, service: EnvCheckService) -> None: + def which(command: str, path: str | None = None) -> str | None: + return "/usr/bin/python" if command == "python" else None with patch( - "shutil.which", - return_value="C:\\nvm4w\\nodejs\\npm.cmd", - ) as mock_which, patch( - "misaka.services.skills.env_check_service.wrap_windows_script_command", - return_value=[ - "cmd.exe", - "/d", - "/s", - "/c", - '"C:\\nvm4w\\nodejs\\npm.cmd" install -g @anthropic-ai/claude-code', - ], - ) as mock_wrap, patch( - "asyncio.create_subprocess_exec", return_value=mock_proc - ) as mock_exec: - result = await service.install_tool("Claude Code CLI") - assert result is True - mock_which.assert_called_once_with("npm", path=ANY) - mock_wrap.assert_called_once_with( - "C:\\nvm4w\\nodejs\\npm.cmd", - ["install", "-g", "@anthropic-ai/claude-code"], + "misaka.services.skills.env_check_service.shutil.which", + side_effect=which, + ), patch.object(service, "_get_version", return_value="3.13.1"): + result = await service._check_tool_multi( + "Python", + ["python3", "python"], + "--version", ) - mock_exec.assert_called_once() - - async def test_install_tool_failure(self, service: EnvCheckService) -> None: - """install_tool should return False on installation failure.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"", b"permission denied\n") - mock_proc.returncode = 1 + assert result.command == "python" + assert result.is_installed is True + async def test_claude_requires_parseable_version(self, service: EnvCheckService) -> None: with patch( - "shutil.which", - return_value="C:\\Windows\\System32\\winget.exe", - ), patch("asyncio.create_subprocess_exec", return_value=mock_proc): - result = await service.install_tool("Node.js") - assert result is False - - async def test_install_tool_unknown_returns_false(self, service: EnvCheckService) -> None: - """install_tool should return False for unknown tools.""" - result = await service.install_tool("UnknownTool") - assert result is False - - async def test_install_tool_missing_launcher_returns_false( - self, service: EnvCheckService - ) -> None: - """install_tool should fail early when launcher cannot be resolved.""" - progress_messages: list[str] = [] - - with patch("shutil.which", return_value=None): - result = await service.install_tool( + "misaka.utils.platform.find_claude_binary", + return_value="/usr/bin/claude", + ), patch( + "misaka.services.skills.env_check_service.shutil.which", + return_value=None, + ), patch.object( + service, + "_get_version", + return_value=None, + ), patch.object( + service, + "_get_version_lenient", + return_value=None, + ): + result = await service._check_tool_multi( "Claude Code CLI", - on_progress=progress_messages.append, - ) - assert result is False - assert any("required command not found: npm" in m for m in progress_messages) - - async def test_install_tool_timeout(self, service: EnvCheckService) -> None: - """install_tool should return False on timeout.""" - mock_proc = AsyncMock() - mock_proc.communicate.side_effect = asyncio.TimeoutError() - - progress_messages: list[str] = [] - - with patch( - "shutil.which", - return_value="C:\\Windows\\System32\\winget.exe", - ), patch("asyncio.create_subprocess_exec", return_value=mock_proc): - result = await service.install_tool( - "Node.js", on_progress=progress_messages.append - ) - assert result is False - assert any("timed out" in m for m in progress_messages) - - async def test_install_tool_os_error(self, service: EnvCheckService) -> None: - """install_tool should return False on OSError (e.g., command not found).""" - progress_messages: list[str] = [] - - with patch( - "shutil.which", - return_value="C:\\Program Files\\Git\\cmd\\git.exe", - ), patch("asyncio.create_subprocess_exec", side_effect=OSError("not found")): - result = await service.install_tool( - "Git", on_progress=progress_messages.append + ["claude"], + "--version", ) - assert result is False - assert any("failed" in m.lower() for m in progress_messages) - - async def test_get_version_no_version_in_output(self, service: EnvCheckService) -> None: - """_get_version should return None when output has no version pattern.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"some random output\n", b"") - mock_proc.returncode = 0 - - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version("/usr/bin/tool", "--version") - assert version is None - - async def test_get_version_os_error(self, service: EnvCheckService) -> None: - """_get_version should return None on OSError.""" - with patch("asyncio.create_subprocess_exec", side_effect=OSError("No such file")): - version = await service._get_version("/nonexistent/path", "--version") - assert version is None - - async def test_get_version_windows_cmd_wrapper(self, service: EnvCheckService) -> None: - """_get_version should delegate .cmd wrappers to the shared command helper.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"v18.0.0\n", b"") - mock_proc.returncode = 0 + assert result.is_installed is False + assert result.version is None + async def test_claude_lenient_version_is_accepted(self, service: EnvCheckService) -> None: with patch( - "misaka.services.skills.env_check_service.wrap_windows_script_command", - return_value=["cmd.exe", "/d", "/s", "/c", '"C:/npm/node.cmd" --version'], - ) as mock_wrap, patch( - "misaka.services.skills.env_check_service.build_background_subprocess_kwargs", - return_value={"creationflags": 1, "startupinfo": "hidden"}, - ) as mock_kwargs, patch( - "asyncio.create_subprocess_exec", return_value=mock_proc - ) as mock_exec: - version = await service._get_version("C:\\npm\\node.cmd", "--version") - assert version == "18.0.0" - mock_wrap.assert_called_once_with("C:\\npm\\node.cmd", ["--version"]) - mock_kwargs.assert_called_once_with() - mock_exec.assert_called_once_with( - "cmd.exe", - "/d", - "/s", - "/c", - '"C:/npm/node.cmd" --version', - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - creationflags=1, - startupinfo="hidden", - ) - - async def test_check_tool_multi_uses_first_found(self, service: EnvCheckService) -> None: - """_check_tool_multi should use the first working command.""" - call_count = 0 - - def mock_which(cmd, path=None): - nonlocal call_count - call_count += 1 - # python3 not found, python found - if cmd == "python3": - return None - return "/usr/bin/python" - - with patch("shutil.which", side_effect=mock_which), \ - patch.object(service, "_get_version", return_value="3.12.1"): + "misaka.utils.platform.find_claude_binary", + return_value="/usr/bin/claude", + ), patch.object( + service, + "_get_version", + return_value=None, + ), patch.object( + service, + "_get_version_lenient", + return_value="2.1.204", + ): result = await service._check_tool_multi( - "Python", ["python3", "python"], "--version" + "Claude Code CLI", + ["claude"], + "--version", ) - assert result.is_installed is True - assert result.command == "python" + assert result.is_installed is True + assert result.version == "2.1.204" - async def test_check_tool_multi_all_missing(self, service: EnvCheckService) -> None: - """_check_tool_multi should return not installed when all commands missing.""" - with patch("shutil.which", return_value=None): - result = await service._check_tool_multi( - "Python", ["python3", "python"], "--version" - ) - assert result.is_installed is False - assert result.command == "python3" # Returns first command as default + async def test_check_all_preserves_tool_name_on_exception( + self, + service: EnvCheckService, + ) -> None: + async def check(name: str, commands: list[str], flag: str) -> ToolStatus: + if name == "Node.js": + raise RuntimeError("boom") + return ToolStatus(name, commands[0], "1.0.0", True, "", "") - async def test_install_tool_no_progress_callback(self, service: EnvCheckService) -> None: - """install_tool should work without a progress callback.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"ok\n", b"") - mock_proc.returncode = 0 + with patch.object(service, "_check_tool_multi", side_effect=check): + result = await service.check_all() - with patch( - "shutil.which", - return_value="C:\\Program Files\\Git\\cmd\\git.exe", - ), patch("asyncio.create_subprocess_exec", return_value=mock_proc): - result = await service.install_tool("Git", on_progress=None) - assert result is True - - async def test_get_version_lenient_ignores_nonzero_exit( - self, service: EnvCheckService, - ) -> None: - """_get_version_lenient should parse version even with non-zero exit code.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"2.1.97 (Claude Code)\n", b"") - mock_proc.returncode = 1 + assert [tool.name for tool in result.tools] == [ + "Claude Code CLI", + "Node.js", + "Python", + "Git", + ] + node = result.tools[1] + assert node.is_installed is False + assert node.install_command + assert result.all_installed is False - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version_lenient("/usr/bin/claude", "--version") - assert version == "2.1.97" - async def test_get_version_lenient_returns_none_on_no_match( - self, service: EnvCheckService, +class TestVersionCapture: + @pytest.mark.parametrize( + ("stdout", "stderr", "expected"), + [ + (b"v24.1.0\n", b"", "24.1.0"), + (b"git version 2.52.0\n", b"", "2.52.0"), + (b"", b"Python 3.13.1\n", "3.13.1"), + (b"2.1.204 (Claude Code)\n", b"", "2.1.204"), + ], + ) + async def test_parses_supported_outputs( + self, + service: EnvCheckService, + stdout: bytes, + stderr: bytes, + expected: str, ) -> None: - """_get_version_lenient returns None when no version pattern found.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"no version here\n", b"") - mock_proc.returncode = 1 + proc = _process(stdout=stdout, stderr=stderr) + with patch("asyncio.create_subprocess_exec", return_value=proc): + assert await service._get_version("/usr/bin/tool", "--version") == expected + + async def test_strict_rejects_nonzero_exit(self, service: EnvCheckService) -> None: + proc = _process(stdout=b"2.1.204\n", returncode=1) + with patch("asyncio.create_subprocess_exec", return_value=proc): + assert await service._get_version("/usr/bin/claude", "--version") is None + + async def test_lenient_accepts_nonzero_exit(self, service: EnvCheckService) -> None: + proc = _process(stdout=b"2.1.204\n", returncode=1) + with patch("asyncio.create_subprocess_exec", return_value=proc): + version = await service._get_version_lenient("/usr/bin/claude", "--version") + assert version == "2.1.204" - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version_lenient("/usr/bin/tool", "--version") - assert version is None + async def test_timeout_terminates_process(self, service: EnvCheckService) -> None: + proc = _process(communicate_error=asyncio.TimeoutError()) + with patch("asyncio.create_subprocess_exec", return_value=proc): + version = await service._get_version("/usr/bin/slow", "--version") + assert version is None + proc.kill.assert_called_once_with() + proc.wait.assert_awaited_once_with() + + +class TestInstallation: + @pytest.fixture(autouse=True) + def successful_verification(self, service: EnvCheckService): + with patch.object( + service, + "_verify_installed_tool", + new=AsyncMock(return_value=True), + ): + yield - async def test_get_version_lenient_returns_none_on_timeout( - self, service: EnvCheckService, + async def test_windows_install_uses_non_interactive_winget( + self, + service: EnvCheckService, ) -> None: - """_get_version_lenient returns None on timeout.""" - with patch("asyncio.create_subprocess_exec", side_effect=asyncio.TimeoutError()): - version = await service._get_version_lenient("/usr/bin/slow", "--version") - assert version is None + proc = _process(stdout=b"installed\n") + progress: list[str] = [] + with patch( + "misaka.services.skills.env_check_service._current_platform", + return_value="windows", + ), patch.object( + service, + "_resolve_install_executable", + return_value=r"C:\Windows\winget.exe", + ), patch( + "asyncio.create_subprocess_exec", + return_value=proc, + ) as create_process: + result = await service.install_tool("Node.js", progress.append) + + assert result.success is True + assert "--accept-source-agreements" in result.command + assert "--disable-interactivity" in result.command + create_process.assert_awaited_once() + command = create_process.await_args.args + assert command[:4] == ( + r"C:\Windows\winget.exe", + "install", + "--id", + "OpenJS.NodeJS.LTS", + ) + assert create_process.await_args.kwargs["stdin"] == asyncio.subprocess.DEVNULL + assert create_process.await_args.kwargs["env"] is not None + assert any("successfully" in message for message in progress) - async def test_check_tool_multi_claude_fallback_to_lenient( - self, service: EnvCheckService, + async def test_macos_install_uses_brew(self, service: EnvCheckService) -> None: + proc = _process() + with patch( + "misaka.services.skills.env_check_service._current_platform", + return_value="macos", + ), patch.object( + service, + "_resolve_install_executable", + return_value="/opt/homebrew/bin/brew", + ), patch("asyncio.create_subprocess_exec", return_value=proc) as create_process: + result = await service.install_tool("Git") + + assert result.success is True + assert create_process.await_args.args == ("/opt/homebrew/bin/brew", "install", "git") + + async def test_linux_apt_runs_update_then_install_as_root( + self, + service: EnvCheckService, ) -> None: - """Claude CLI check falls back to lenient version when strict fails.""" + processes = [_process(), _process()] with patch( - "misaka.utils.platform.find_claude_binary", - return_value="/usr/bin/claude", + "misaka.services.skills.env_check_service._current_platform", + return_value="linux", ), patch.object( - service, "_get_version", return_value=None, + service, + "_resolve_install_executable", + return_value="/usr/bin/apt-get", ), patch.object( - service, "_get_version_lenient", return_value="2.1.97", - ): - result = await service._check_tool_multi( - "Claude Code CLI", ["claude"], "--version", - ) - assert result.is_installed is True - assert result.version == "2.1.97" + service, + "_is_root", + return_value=True, + ), patch( + "asyncio.create_subprocess_exec", + side_effect=processes, + ) as create_process: + result = await service.install_tool("Git") + + assert result.success is True + assert create_process.await_args_list[0].args[:2] == ("/usr/bin/apt-get", "update") + assert create_process.await_args_list[1].args[:4] == ( + "/usr/bin/apt-get", + "install", + "-y", + "git", + ) - async def test_check_tool_multi_claude_installed_no_version( - self, service: EnvCheckService, + async def test_linux_apt_uses_noninteractive_elevation( + self, + service: EnvCheckService, ) -> None: - """Claude CLI is_installed=True even when both version methods return None.""" + proc = _process() with patch( - "misaka.utils.platform.find_claude_binary", - return_value="/usr/bin/claude", - ), patch.object( - service, "_get_version", return_value=None, + "misaka.services.skills.env_check_service._current_platform", + return_value="linux", ), patch.object( - service, "_get_version_lenient", return_value=None, - ): - result = await service._check_tool_multi( - "Claude Code CLI", ["claude"], "--version", - ) - assert result.is_installed is True - assert result.version is None + service, + "_resolve_install_executable", + return_value="/usr/bin/apt-get", + ), patch.object(service, "_is_root", return_value=False), patch.object( + service, + "_resolve_elevation_command", + return_value=["/usr/bin/sudo", "--non-interactive"], + ), patch("asyncio.create_subprocess_exec", return_value=proc) as create_process: + result = await service.install_tool("Python") + + assert result.success is True + assert create_process.await_args_list[0].args[:4] == ( + "/usr/bin/sudo", + "--non-interactive", + "/usr/bin/apt-get", + "update", + ) - async def test_get_version_parses_claude_output( - self, service: EnvCheckService, + async def test_missing_launcher_returns_actionable_error( + self, + service: EnvCheckService, ) -> None: - """_get_version should parse Claude Code CLI output format.""" - mock_proc = AsyncMock() - mock_proc.communicate.return_value = (b"2.1.97 (Claude Code)\n", b"") - mock_proc.returncode = 0 - - with patch("asyncio.create_subprocess_exec", return_value=mock_proc): - version = await service._get_version("/usr/bin/claude", "--version") - assert version == "2.1.97" + with patch( + "misaka.services.skills.env_check_service._current_platform", + return_value="macos", + ), patch.object(service, "_resolve_install_executable", return_value=None): + result = await service.install_tool("Node.js") + assert result.success is False + assert "brew" in result.message + assert "nodejs.org" in result.message + + async def test_failure_returns_stderr_and_code(self, service: EnvCheckService) -> None: + proc = _process(stderr=b"permission denied\n", returncode=1) + with patch.object( + service, + "_resolve_install_executable", + return_value=r"C:\Windows\winget.exe", + ), patch("asyncio.create_subprocess_exec", return_value=proc): + result = await service.install_tool("Git") + assert result.success is False + assert result.returncode == 1 + assert "permission denied" in result.message + + async def test_zero_exit_still_requires_detectable_tool( + self, + service: EnvCheckService, + ) -> None: + proc = _process() + with patch.object( + service, + "_resolve_install_executable", + return_value=r"C:\Windows\winget.exe", + ), patch.object( + service, + "_verify_installed_tool", + new=AsyncMock(return_value=False), + ), patch("asyncio.create_subprocess_exec", return_value=proc): + result = await service.install_tool("Git") + + assert result.success is False + assert result.returncode == 0 + assert "could not be detected" in result.message + + async def test_progress_callback_failure_does_not_abort_install( + self, + service: EnvCheckService, + ) -> None: + proc = _process() + progress = MagicMock(side_effect=RuntimeError("detached UI")) + with patch.object( + service, + "_resolve_install_executable", + return_value=r"C:\Windows\winget.exe", + ), patch("asyncio.create_subprocess_exec", return_value=proc): + result = await service.install_tool("Git", progress) + + assert result.success is True + + async def test_timeout_kills_process(self, service: EnvCheckService) -> None: + proc = _process(communicate_error=asyncio.TimeoutError()) + with patch.object( + service, + "_resolve_install_executable", + return_value=r"C:\Windows\winget.exe", + ), patch("asyncio.create_subprocess_exec", return_value=proc): + result = await service.install_tool("Git") + assert result.success is False + assert "timed out" in result.message + proc.kill.assert_called_once_with() + proc.wait.assert_awaited_once_with() + + async def test_unknown_tool_returns_failure(self, service: EnvCheckService) -> None: + result = await service.install_tool("Unknown") + assert result.success is False + assert result.command == "" + + async def test_subprocess_receives_hidden_window_kwargs( + self, + service: EnvCheckService, + ) -> None: + proc = _process() + with patch( + "misaka.services.skills.env_check_service._current_platform", + return_value="windows", + ), patch.object( + service, + "_resolve_install_executable", + return_value=r"C:\Windows\winget.exe", + ), patch( + "misaka.services.skills.env_check_service.build_background_subprocess_kwargs", + return_value={"creationflags": 1, "startupinfo": "hidden"}, + ), patch("asyncio.create_subprocess_exec", return_value=proc) as create_process: + await service.install_tool("Git") + create_process.assert_awaited_once() + kwargs = create_process.await_args.kwargs + assert kwargs["stdin"] == asyncio.subprocess.DEVNULL + assert kwargs["stdout"] == asyncio.subprocess.PIPE + assert kwargs["stderr"] == asyncio.subprocess.PIPE + assert kwargs["env"] is not None + assert kwargs["creationflags"] == 1 + assert kwargs["startupinfo"] == "hidden" diff --git a/tests/unit/test_env_check_ui.py b/tests/unit/test_env_check_ui.py new file mode 100644 index 0000000..d79bc98 --- /dev/null +++ b/tests/unit/test_env_check_ui.py @@ -0,0 +1,70 @@ +"""UI state tests for environment install progress and errors.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import misaka.i18n as i18n +from misaka.services.skills.env_check_service import ( + EnvironmentCheckResult, + InstallResult, + ToolStatus, +) +from misaka.ui.dialogs.env_check_dialog import EnvCheckDialog +from misaka.ui.settings.components.env_status_panel import EnvStatusPanel + + +def _state() -> MagicMock: + state = MagicMock() + state.env_check_result = EnvironmentCheckResult( + tools=[ + ToolStatus( + "Git", + "git", + None, + False, + "https://git-scm.com/downloads", + "winget install Git.Git", + ) + ], + all_installed=False, + checked_at="2026-08-11T00:00:00Z", + ) + return state + + +def setup_module() -> None: + i18n.init("en") + + +def test_dialog_retains_install_error() -> None: + dialog = EnvCheckDialog(_state()) + result = InstallResult("Git", False, "permission denied", returncode=1) + + dialog.finish_install(result) + + assert dialog._installing_tool is None + assert dialog._status_is_error is True + assert dialog._status_message is not None + assert "permission denied" in dialog._status_message + + +def test_dialog_retains_install_success() -> None: + dialog = EnvCheckDialog(_state()) + result = InstallResult("Git", True, "installed") + + dialog.finish_install(result) + + assert dialog._status_is_error is False + assert dialog._status_message == "Git installed successfully." + + +def test_settings_panel_retains_install_error() -> None: + panel = EnvStatusPanel(_state()) + result = InstallResult("Git", False, "permission denied", returncode=1) + + panel._finish_install(result) + + assert panel._install_error is True + assert panel._install_message is not None + assert "permission denied" in panel._install_message diff --git a/tests/unit/test_update_check_service.py b/tests/unit/test_update_check_service.py index 107bfd6..7af8c1c 100644 --- a/tests/unit/test_update_check_service.py +++ b/tests/unit/test_update_check_service.py @@ -134,7 +134,10 @@ async def test_http_get_version_failure(self, service: UpdateCheckService) -> No """_http_get_version should return None on network failure.""" from urllib.error import URLError - with patch("misaka.services.file.update_check_service.urlopen", side_effect=URLError("fail")): + with patch( + "misaka.services.file.update_check_service.urlopen", + side_effect=URLError("fail"), + ): version = service._http_get_version() assert version is None @@ -157,8 +160,11 @@ async def test_perform_update_success(self, service: UpdateCheckService) -> None progress_messages: list[str] = [] - with patch("shutil.which", return_value="C:/npm/npm.cmd"), \ - patch( + with patch.object( + service, + "_resolve_update_command", + return_value=["C:/npm/npm.cmd", "install", "-g", "package"], + ), patch( "misaka.services.file.update_check_service.build_background_subprocess_kwargs", return_value={"creationflags": 1, "startupinfo": "hidden"}, ) as mock_kwargs, \ @@ -167,25 +173,23 @@ async def test_perform_update_success(self, service: UpdateCheckService) -> None result = await service.perform_update(on_progress=progress_messages.append) assert result is True mock_kwargs.assert_called_once_with() - mock_exec.assert_called_once_with( - "C:/npm/npm.cmd", - "install", - "-g", - "@anthropic-ai/claude-code@latest", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - creationflags=1, - startupinfo="hidden", + mock_exec.assert_awaited_once() + assert mock_exec.await_args.args == ( + "C:/npm/npm.cmd", "install", "-g", "package" ) + assert mock_exec.await_args.kwargs["stdin"] == asyncio.subprocess.DEVNULL + assert mock_exec.await_args.kwargs["env"] is not None + assert mock_exec.await_args.kwargs["creationflags"] == 1 + assert mock_exec.await_args.kwargs["startupinfo"] == "hidden" - async def test_perform_update_no_npm(self, service: UpdateCheckService) -> None: - """perform_update should return False when npm is not found.""" + async def test_perform_update_no_manager(self, service: UpdateCheckService) -> None: + """perform_update should return False when no manager is found.""" progress_messages: list[str] = [] - with patch("shutil.which", return_value=None): + with patch.object(service, "_resolve_update_command", return_value=None): result = await service.perform_update(on_progress=progress_messages.append) assert result is False - assert any("npm not found" in m for m in progress_messages) + assert any("manager" in m for m in progress_messages) async def test_perform_update_failure(self, service: UpdateCheckService) -> None: """perform_update should return False on update failure.""" @@ -193,8 +197,9 @@ async def test_perform_update_failure(self, service: UpdateCheckService) -> None mock_proc.communicate.return_value = (b"", b"error occurred\n") mock_proc.returncode = 1 - with patch("shutil.which", return_value="/usr/bin/npm"), \ - patch("asyncio.create_subprocess_exec", return_value=mock_proc): + with patch.object( + service, "_resolve_update_command", return_value=["claude", "update"] + ), patch("asyncio.create_subprocess_exec", return_value=mock_proc): result = await service.perform_update() assert result is False @@ -241,7 +246,10 @@ async def test_fetch_version_via_npm_cli_bad_output(self, service: UpdateCheckSe version = await service._fetch_version_via_npm_cli("@anthropic-ai/claude-code") assert version is None - async def test_fetch_version_via_npm_cli_failure_exit(self, service: UpdateCheckService) -> None: + async def test_fetch_version_via_npm_cli_failure_exit( + self, + service: UpdateCheckService, + ) -> None: """_fetch_version_via_npm_cli should return None on non-zero exit.""" mock_proc = AsyncMock() mock_proc.communicate.return_value = (b"", b"error\n") @@ -254,27 +262,100 @@ async def test_fetch_version_via_npm_cli_failure_exit(self, service: UpdateCheck async def test_perform_update_timeout(self, service: UpdateCheckService) -> None: """perform_update should return False on timeout.""" - mock_proc = AsyncMock() - mock_proc.communicate.side_effect = asyncio.TimeoutError() + mock_proc = MagicMock() + mock_proc.communicate = AsyncMock(side_effect=asyncio.TimeoutError()) + mock_proc.wait = AsyncMock() + mock_proc.kill = MagicMock() progress_messages: list[str] = [] - with patch("shutil.which", return_value="/usr/bin/npm"), \ - patch("asyncio.create_subprocess_exec", return_value=mock_proc): + with patch.object( + service, "_resolve_update_command", return_value=["claude", "update"] + ), patch("asyncio.create_subprocess_exec", return_value=mock_proc): result = await service.perform_update(on_progress=progress_messages.append) assert result is False assert any("timed out" in m.lower() for m in progress_messages) + mock_proc.kill.assert_called_once_with() + mock_proc.wait.assert_awaited_once_with() async def test_perform_update_exception(self, service: UpdateCheckService) -> None: """perform_update should return False on unexpected exception.""" progress_messages: list[str] = [] - with patch("shutil.which", return_value="/usr/bin/npm"), \ - patch("asyncio.create_subprocess_exec", side_effect=OSError("broken")): + with patch.object( + service, "_resolve_update_command", return_value=["claude", "update"] + ), patch("asyncio.create_subprocess_exec", side_effect=OSError("broken")): result = await service.perform_update(on_progress=progress_messages.append) assert result is False assert any("failed" in m.lower() for m in progress_messages) + def test_resolve_update_command_for_npm_install( + self, service: UpdateCheckService + ) -> None: + with patch( + "misaka.utils.platform.find_claude_binary", + return_value="C:/npm/claude.cmd", + ), patch( + "misaka.services.file.update_check_service.shutil.which", + return_value="C:/npm/npm.cmd", + ): + command = service._resolve_update_command() + assert command == [ + "C:/npm/npm.cmd", + "install", + "-g", + "@anthropic-ai/claude-code@latest", + ] + + def test_resolve_update_command_for_winget_install( + self, service: UpdateCheckService + ) -> None: + with patch( + "misaka.utils.platform.find_claude_binary", + return_value="C:/Users/test/AppData/Local/Microsoft/WinGet/Links/claude.exe", + ), patch( + "misaka.services.file.update_check_service.IS_WINDOWS", + True, + ), patch( + "misaka.services.file.update_check_service.shutil.which", + return_value="C:/Windows/winget.exe", + ): + command = service._resolve_update_command() + assert command is not None + assert command[:4] == [ + "C:/Windows/winget.exe", + "upgrade", + "--id", + "Anthropic.ClaudeCode", + ] + assert "--disable-interactivity" in command + + def test_resolve_update_command_for_homebrew_install( + self, service: UpdateCheckService + ) -> None: + with patch( + "misaka.utils.platform.find_claude_binary", + return_value="/opt/homebrew/Caskroom/claude-code/2.1.204/claude", + ), patch( + "misaka.services.file.update_check_service.IS_MACOS", + True, + ), patch( + "misaka.services.file.update_check_service.shutil.which", + return_value="/opt/homebrew/bin/brew", + ): + command = service._resolve_update_command() + assert command == ["/opt/homebrew/bin/brew", "upgrade", "claude-code"] + + def test_resolve_update_command_for_native_install( + self, service: UpdateCheckService + ) -> None: + with patch( + "misaka.utils.platform.find_claude_binary", + return_value="/home/test/.local/bin/claude", + ): + command = service._resolve_update_command() + assert command == ["/home/test/.local/bin/claude", "update"] + async def test_http_get_version_missing_version_key(self, service: UpdateCheckService) -> None: """_http_get_version should return None when JSON has no version key.""" mock_response = MagicMock() @@ -286,7 +367,10 @@ async def test_http_get_version_missing_version_key(self, service: UpdateCheckSe version = service._http_get_version() assert version is None - async def test_http_get_version_invalid_version_format(self, service: UpdateCheckService) -> None: + async def test_http_get_version_invalid_version_format( + self, + service: UpdateCheckService, + ) -> None: """_http_get_version should return None when version doesn't match pattern.""" mock_response = MagicMock() mock_response.read.return_value = json.dumps({"version": "not-a-version"}).encode() @@ -299,12 +383,17 @@ async def test_http_get_version_invalid_version_format(self, service: UpdateChec async def test_get_current_version_no_claude(self, service: UpdateCheckService) -> None: """_get_current_version should return None when claude CLI not found.""" - with patch("misaka.services.file.update_check_service.find_claude_binary", return_value=None, create=True), \ - patch.dict("sys.modules", {}): - # Patch the import chain - with patch.object(service, "_get_current_version", return_value=None): - result = await service.check_for_update() - assert result.current_version is None + with patch( + "misaka.services.file.update_check_service.find_claude_binary", + return_value=None, + create=True, + ), patch.dict("sys.modules", {}), patch.object( + service, + "_get_current_version", + return_value=None, + ): + result = await service.check_for_update() + assert result.current_version is None async def test_check_for_update_both_none(self, service: UpdateCheckService) -> None: """check_for_update should handle both versions being None."""