From 32a52bcfe7b9fb0f9d4ee5c14a87a8b1ace04ade Mon Sep 17 00:00:00 2001 From: sahil Date: Sun, 7 Dec 2025 17:46:41 +0530 Subject: [PATCH 1/3] Installation simulation mode fix #103 --- cortex/cli.py | 35 +- cortex/preflight_checker.py | 605 +++++++++++++++++++++++++++++++++ docs/SIMULATION_MODE.md | 228 +++++++++++++ requirements.txt | 1 + test/test_cli.py | 6 +- test/test_preflight_checker.py | 201 +++++++++++ 6 files changed, 1071 insertions(+), 5 deletions(-) create mode 100644 cortex/preflight_checker.py create mode 100644 docs/SIMULATION_MODE.md create mode 100644 test/test_preflight_checker.py diff --git a/cortex/cli.py b/cortex/cli.py index b3981a9..1a4d6bd 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -20,6 +20,7 @@ print_all_preferences, format_preference_value ) +from cortex.preflight_checker import PreflightChecker, format_report, export_report class CortexCLI: @@ -61,7 +62,11 @@ def _clear_line(self): sys.stdout.write('\r\033[K') sys.stdout.flush() - def install(self, software: str, execute: bool = False, dry_run: bool = False): + def install(self, software: str, execute: bool = False, dry_run: bool = False, simulate: bool = False): + # Handle simulation mode first - no API key needed + if simulate: + return self._run_simulation(software) + api_key = self._get_api_key() if not api_key: return 1 @@ -187,6 +192,30 @@ def progress_callback(current, total, step): history.update_installation(install_id, InstallationStatus.FAILED, str(e)) self._print_error(f"Unexpected error: {str(e)}") return 1 + + def _run_simulation(self, software: str) -> int: + """Run preflight simulation check for installation""" + try: + # Get API key for LLM-powered package info (optional) + api_key = self._get_api_key() + provider = self._get_provider() if api_key else 'openai' + + # Create checker with optional API key for enhanced accuracy + checker = PreflightChecker(api_key=api_key, provider=provider) + report = checker.run_all_checks(software) + + # Print formatted report + output = format_report(report, software) + print(output) + + # Return error code if blocking issues found + if report.errors: + return 1 + return 0 + + except Exception as e: + self._print_error(f"Simulation failed: {str(e)}") + return 1 def history(self, limit: int = 20, status: Optional[str] = None, show_id: Optional[str] = None): """Show installation history""" @@ -496,6 +525,7 @@ def main(): Examples: cortex install docker cortex install docker --execute + cortex install docker --simulate cortex install "python 3.11 with pip" cortex install nginx --dry-run cortex history @@ -520,6 +550,7 @@ def main(): install_parser.add_argument('software', type=str, help='Software to install (natural language)') install_parser.add_argument('--execute', action='store_true', help='Execute the generated commands') install_parser.add_argument('--dry-run', action='store_true', help='Show commands without executing') + install_parser.add_argument('--simulate', action='store_true', help='Simulate installation without making changes') # History command history_parser = subparsers.add_parser('history', help='View installation history') @@ -556,7 +587,7 @@ def main(): try: if args.command == 'install': - return cli.install(args.software, execute=args.execute, dry_run=args.dry_run) + return cli.install(args.software, execute=args.execute, dry_run=args.dry_run, simulate=args.simulate) elif args.command == 'history': return cli.history(limit=args.limit, status=args.status, show_id=args.show_id) elif args.command == 'rollback': diff --git a/cortex/preflight_checker.py b/cortex/preflight_checker.py new file mode 100644 index 0000000..e1f8c4c --- /dev/null +++ b/cortex/preflight_checker.py @@ -0,0 +1,605 @@ +""" +Preflight System Checker for Cortex Installation + +Performs real system checks before installation to identify issues, +verify requirements, and predict what will be installed. +""" + +import os +import sys +import shutil +import socket +import platform +import subprocess +import json +from dataclasses import dataclass, field +from typing import List, Dict, Optional, Tuple +from pathlib import Path + + +@dataclass +class DiskInfo: + """Disk usage information for a path""" + path: str + free_mb: int + total_mb: int + filesystem: str + exists: bool + writable: bool + + +@dataclass +class PackageInfo: + """Information about a package/binary""" + name: str + installed: bool + version: Optional[str] = None + path: Optional[str] = None + + +@dataclass +class ServiceInfo: + """Information about a system service""" + name: str + exists: bool + active: bool + enabled: bool = False + + +@dataclass +class PreflightReport: + """Complete preflight check report""" + os_info: Dict[str, str] = field(default_factory=dict) + kernel_info: Dict[str, str] = field(default_factory=dict) + cpu_arch: str = "" + cgroup_info: Dict[str, str] = field(default_factory=dict) + disk_usage: List[DiskInfo] = field(default_factory=list) + package_status: List[PackageInfo] = field(default_factory=list) + service_status: List[ServiceInfo] = field(default_factory=list) + permissions_status: Dict[str, bool] = field(default_factory=dict) + network_status: Dict[str, bool] = field(default_factory=dict) + errors: List[str] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + suggestions: List[str] = field(default_factory=list) + + # Installation estimates based on real checks + packages_to_install: List[Dict[str, str]] = field(default_factory=list) + total_download_mb: int = 0 + total_disk_required_mb: int = 0 + config_changes: List[str] = field(default_factory=list) + + +class PreflightChecker: + """ + Real system checker for Cortex installation preflight checks. + + All checks are performed against the actual system - no simulation. + """ + + # Paths to check for disk space + DISK_CHECK_PATHS = ['/', '/var/lib/docker', '/opt'] + MIN_DISK_SPACE_MB = 500 + + def __init__(self, api_key: Optional[str] = None, provider: str = 'openai'): + self.report = PreflightReport() + self._is_linux = sys.platform.startswith('linux') + self._is_windows = sys.platform == 'win32' + self._is_mac = sys.platform == 'darwin' + self.api_key = api_key + self.provider = provider + self._llm_client = None + + def _run_command(self, cmd: List[str], timeout: int = 10) -> Tuple[bool, str]: + """Run a shell command and return success status and output""" + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout + ) + return result.returncode == 0, result.stdout.strip() + except subprocess.TimeoutExpired: + return False, "Command timed out" + except FileNotFoundError: + return False, "Command not found" + except Exception as e: + return False, str(e) + + def _run_shell_command(self, cmd: str, timeout: int = 10) -> Tuple[bool, str]: + """Run a shell command string and return success status and output""" + try: + result = subprocess.run( + cmd, + shell=True, + capture_output=True, + text=True, + timeout=timeout + ) + return result.returncode == 0, result.stdout.strip() + except subprocess.TimeoutExpired: + return False, "Command timed out" + except Exception as e: + return False, str(e) + + def check_os_info(self) -> Dict[str, str]: + """Detect OS and distribution information""" + info = { + 'platform': platform.system(), + 'platform_release': platform.release(), + 'platform_version': platform.version(), + 'machine': platform.machine(), + } + + if self._is_linux: + # Try to get distro info + distro_info = self._get_linux_distro() + info.update(distro_info) + elif self._is_windows: + info['distro'] = 'Windows' + info['distro_version'] = platform.version() + elif self._is_mac: + info['distro'] = 'macOS' + success, version = self._run_command(['sw_vers', '-productVersion']) + info['distro_version'] = version if success else platform.mac_ver()[0] + + self.report.os_info = info + return info + + def _get_linux_distro(self) -> Dict[str, str]: + """Get Linux distribution information from /etc/os-release""" + info = {'distro': 'Unknown', 'distro_version': '', 'distro_id': ''} + + os_release_path = Path('/etc/os-release') + if os_release_path.exists(): + try: + with open(os_release_path, 'r') as f: + for line in f: + line = line.strip() + if '=' in line: + key, value = line.split('=', 1) + value = value.strip('"\'') + if key == 'NAME': + info['distro'] = value + elif key == 'VERSION_ID': + info['distro_version'] = value + elif key == 'ID': + info['distro_id'] = value + except Exception: + pass + + return info + + def check_basic_system_info(self) -> Dict[str, str]: + """Get basic system information for display""" + info = { + 'kernel': platform.release(), + 'architecture': platform.machine() + } + + self.report.kernel_info = {'version': info['kernel']} + self.report.cpu_arch = info['architecture'] + + return info + + def check_disk_space(self, additional_paths: Optional[List[str]] = None) -> List[DiskInfo]: + """Check disk space on critical paths""" + paths_to_check = list(self.DISK_CHECK_PATHS) + + # Add current working directory + paths_to_check.append(os.getcwd()) + + if additional_paths: + paths_to_check.extend(additional_paths) + + # Remove duplicates while preserving order + seen = set() + unique_paths = [] + for p in paths_to_check: + if p not in seen: + seen.add(p) + unique_paths.append(p) + + disk_info_list = [] + + for path in unique_paths: + disk_info = self._get_disk_info(path) + disk_info_list.append(disk_info) + + # Check if path has enough space + if disk_info.exists and disk_info.free_mb < self.MIN_DISK_SPACE_MB: + self.report.warnings.append( + f"Low disk space on {path}: {disk_info.free_mb} MB free" + ) + + self.report.disk_usage = disk_info_list + return disk_info_list + + def _get_disk_info(self, path: str) -> DiskInfo: + """Get disk usage information for a specific path""" + exists = os.path.exists(path) + writable = os.access(path, os.W_OK) if exists else False + + if not exists: + # Try to find the nearest existing parent + check_path = Path(path) + while not check_path.exists() and check_path.parent != check_path: + check_path = check_path.parent + path = str(check_path) + exists = check_path.exists() + + free_mb = 0 + total_mb = 0 + filesystem = 'unknown' + + if exists: + try: + stat = shutil.disk_usage(path) + free_mb = stat.free // (1024 * 1024) + total_mb = stat.total // (1024 * 1024) + except Exception: + pass + + # Get filesystem type on Linux + if self._is_linux: + filesystem = self._get_filesystem_type(path) + + return DiskInfo( + path=path, + free_mb=free_mb, + total_mb=total_mb, + filesystem=filesystem, + exists=exists, + writable=writable + ) + + def _get_filesystem_type(self, path: str) -> str: + """Get filesystem type for a path on Linux""" + success, output = self._run_shell_command(f"df -T '{path}' | tail -1 | awk '{{print $2}}'") + if success and output: + return output + return 'unknown' + + + + def check_package(self, name: str, version_cmd: Optional[List[str]] = None) -> PackageInfo: + """Check if a package/binary is installed""" + # First check if binary exists in PATH + path = shutil.which(name) + installed = path is not None + version = None + + if installed and version_cmd: + success, output = self._run_command(version_cmd) + if success: + version = output.split('\n')[0] + + return PackageInfo( + name=name, + installed=installed, + version=version, + path=path + ) + + def _get_llm_client(self): + """Lazy initialize LLM client with fallback""" + if self._llm_client is not None: + return self._llm_client + + if not self.api_key: + return None + + # Try to initialize with primary provider + try: + sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + from LLM.interpreter import CommandInterpreter + self._llm_client = CommandInterpreter(api_key=self.api_key, provider=self.provider) + return self._llm_client + except Exception: + # Fallback to other provider + try: + fallback_provider = 'claude' if self.provider == 'openai' else 'openai' + # Get fallback API key + fallback_key = os.environ.get('ANTHROPIC_API_KEY' if fallback_provider == 'claude' else 'OPENAI_API_KEY') + if fallback_key: + from LLM.interpreter import CommandInterpreter + self._llm_client = CommandInterpreter(api_key=fallback_key, provider=fallback_provider) + return self._llm_client + except Exception: + pass + + return None + + def _get_package_info_from_llm(self, software: str, os_info: Dict[str, str]) -> Dict[str, any]: + """Query LLM for real package information including sizes""" + client = self._get_llm_client() + if not client: + return {'packages': [], 'total_size_mb': 0, 'config_changes': []} + + try: + # Create a specific prompt for package information + distro = os_info.get('distro', 'Ubuntu') + distro_id = os_info.get('distro_id', 'ubuntu') + + prompt = f"""Get {software} package information for Linux {distro_id}. + +RESPOND WITH ONLY JSON - NO EXPLANATIONS OR TEXT BEFORE/AFTER: +{{ + "packages": [ + {{"name": "exact-package-name", "version": "latest", "size_mb": 25}} + ], + "total_size_mb": 25, + "config_changes": ["/etc/nginx/nginx.conf"] +}} + +Provide real package sizes in MB (integers only). Use standard repository package names for {distro_id}. +If unsure, estimate typical sizes. ONLY OUTPUT THE JSON OBJECT.""" + + # Use the LLM's API directly for a simpler query + if hasattr(client, 'client'): + if client.provider.value == 'openai': + response = client.client.chat.completions.create( + model=client.model, + messages=[ + {"role": "system", "content": "You are a Linux package expert. Provide accurate package information in JSON format only."}, + {"role": "user", "content": prompt} + ], + temperature=0.1, + max_tokens=1000 + ) + content = response.choices[0].message.content.strip() + else: # claude + response = client.client.messages.create( + model=client.model, + max_tokens=1000, + temperature=0.1, + system="You are a Linux package expert. Provide accurate package information in JSON format only.", + messages=[{"role": "user", "content": prompt}] + ) + content = response.content[0].text.strip() + + # Parse JSON response + if content.startswith("```json"): + content = content.split("```json")[1].split("```")[0].strip() + elif content.startswith("```"): + content = content.split("```")[1].split("```")[0].strip() + + data = json.loads(content) + return data + except Exception: + # LLM response parsing failed, fall back to estimates + pass + + return {'packages': [], 'total_size_mb': 0, 'config_changes': []} + + def check_docker(self) -> PackageInfo: + """Check Docker installation and version""" + pkg = self.check_package('docker', ['docker', '--version']) + self.report.package_status.append(pkg) + + if not pkg.installed: + # Try to get real package info from LLM + if self.api_key: + pkg_info = self._get_package_info_from_llm('docker', self.report.os_info) + if pkg_info['packages']: + self.report.packages_to_install.extend(pkg_info['packages']) + self.report.config_changes.extend(pkg_info.get('config_changes', [])) + else: + # Fallback to defaults if LLM fails + self.report.packages_to_install.append({ + 'name': 'docker-ce', + 'version': 'latest', + 'size_mb': '~85 (estimate)' + }) + self.report.config_changes.append('/etc/docker/daemon.json (new file)') + self.report.config_changes.append('/etc/group (add docker group)') + else: + # No API key - show estimate with disclaimer + self.report.packages_to_install.append({ + 'name': 'docker-ce', + 'version': 'latest', + 'size_mb': '~85 (estimate)' + }) + self.report.config_changes.append('/etc/docker/daemon.json (new file)') + self.report.config_changes.append('/etc/group (add docker group)') + self.report.warnings.append('Package sizes are estimates - provide API key for accurate sizes') + + return pkg + + def check_containerd(self) -> PackageInfo: + """Check containerd installation""" + pkg = self.check_package('containerd', ['containerd', '--version']) + self.report.package_status.append(pkg) + + if not pkg.installed: + # Try to get real package info from LLM + if self.api_key: + pkg_info = self._get_package_info_from_llm('containerd', self.report.os_info) + if pkg_info['packages']: + self.report.packages_to_install.extend(pkg_info['packages']) + else: + self.report.packages_to_install.append({ + 'name': 'containerd.io', + 'version': 'latest', + 'size_mb': '~45 (estimate)' + }) + else: + self.report.packages_to_install.append({ + 'name': 'containerd.io', + 'version': 'latest', + 'size_mb': '~45 (estimate)' + }) + + return pkg + + def check_software(self, software_name: str) -> PackageInfo: + """Check any software installation dynamically""" + # Try common binary names + pkg = self.check_package(software_name.lower(), [software_name.lower(), '--version']) + self.report.package_status.append(pkg) + + if not pkg.installed: + # Try to get real package info from LLM + if self.api_key: + pkg_info = self._get_package_info_from_llm(software_name, self.report.os_info) + if pkg_info.get('packages'): + self.report.packages_to_install.extend(pkg_info['packages']) + if pkg_info.get('config_changes'): + self.report.config_changes.extend(pkg_info['config_changes']) + else: + # LLM didn't return data, use estimate + self.report.packages_to_install.append({ + 'name': software_name, + 'version': 'latest', + 'size_mb': '~50 (estimate)' + }) + self.report.warnings.append(f'Could not fetch real package size for {software_name}') + else: + # No API key, use estimate + self.report.packages_to_install.append({ + 'name': software_name, + 'version': 'latest', + 'size_mb': '~50 (estimate)' + }) + + return pkg + + def calculate_requirements(self, software: str) -> None: + """Calculate installation requirements based on software to install""" + software_lower = software.lower() + + # Calculate total download and disk requirements + total_download = 0 + total_disk = 0 + + for pkg in self.report.packages_to_install: + try: + size_str = str(pkg.get('size_mb', '0')) + # Remove estimate markers and parse + size_str = size_str.replace('~', '').replace('(estimate)', '').strip() + size = int(float(size_str)) + total_download += size + total_disk += size * 3 # Rough estimate: downloaded + extracted + working space + except (ValueError, AttributeError): + pass + + self.report.total_download_mb = total_download + self.report.total_disk_required_mb = total_disk + + # Check if we have enough disk space + root_disk = next( + (d for d in self.report.disk_usage if d.path == '/'), + None + ) + + if root_disk and root_disk.free_mb < total_disk: + self.report.errors.append( + f"Insufficient disk space: {root_disk.free_mb} MB available, {total_disk} MB required" + ) + + def run_all_checks(self, software: str = "docker") -> PreflightReport: + """Run all preflight checks and return complete report""" + + # OS and system detection + self.check_os_info() + self.check_basic_system_info() + + # Disk checks + self.check_disk_space() + + # Package checks - check the requested software + software_lower = software.lower() + + if 'docker' in software_lower: + self.check_docker() + self.check_containerd() + else: + # Generic software check with LLM + self.check_software(software) + + # Calculate requirements + self.calculate_requirements(software) + + return self.report + + +def format_report(report: PreflightReport, software: str) -> str: + """Format the preflight report for display""" + lines = [] + lines.append(f"\nšŸ” Simulation mode: No changes will be made\n") + + # Check if using estimates + using_estimates = any('estimate' in str(pkg.get('size_mb', '')) for pkg in report.packages_to_install) + + # System info + lines.append("System Information:") + lines.append(f" OS: {report.os_info.get('distro', 'Unknown')} {report.os_info.get('distro_version', '')}") + if report.kernel_info.get('version'): + lines.append(f" Kernel: {report.kernel_info.get('version')}") + if report.cpu_arch: + lines.append(f" Architecture: {report.cpu_arch}") + + # What would be installed + if report.packages_to_install: + lines.append(f"\nWould install:") + for pkg in report.packages_to_install: + lines.append(f" - {pkg['name']} {pkg.get('version', '')} ({pkg.get('size_mb', '?')} MB)") + + if using_estimates: + lines.append(f"\nTotal download: ~{report.total_download_mb} MB (estimate)") + lines.append(f"Disk space required: ~{report.total_disk_required_mb} MB (estimate)") + lines.append("\nšŸ’” Tip: Set OPENAI_API_KEY or ANTHROPIC_API_KEY for real-time package sizes") + else: + lines.append(f"\nTotal download: {report.total_download_mb} MB") + lines.append(f"Disk space required: {report.total_disk_required_mb} MB") + else: + lines.append(f"\nāœ“ {software} is already installed") + + # Disk space available + root_disk = next((d for d in report.disk_usage if d.path == '/'), None) + if root_disk: + status = 'āœ“' if root_disk.free_mb > report.total_disk_required_mb else 'āœ—' + lines.append(f"Disk space available: {root_disk.free_mb // 1024} GB {status}") + + # Configuration changes + if report.config_changes: + lines.append("\nWould modify:") + for change in report.config_changes: + lines.append(f" - {change}") + + # Potential issues + if report.errors: + lines.append("\nāŒ Blocking issues:") + for error in report.errors: + lines.append(f" - {error}") + elif report.warnings: + lines.append("\nāš ļø Warnings:") + for warning in report.warnings[:5]: # Show first 5 warnings + lines.append(f" - {warning}") + else: + lines.append("\nPotential issues: None detected") + + # Suggestions + if report.suggestions: + lines.append("\nšŸ’” Suggestions:") + for suggestion in report.suggestions[:3]: + lines.append(f" - {suggestion}") + + return '\n'.join(lines) + + +def export_report(report: PreflightReport, filepath: str) -> None: + """Export preflight report to a JSON file""" + import json + from dataclasses import asdict + + # Convert dataclass to dict + report_dict = asdict(report) + + # Convert DiskInfo, PackageInfo, ServiceInfo to dicts + report_dict['disk_usage'] = [asdict(d) for d in report.disk_usage] + report_dict['package_status'] = [asdict(p) for p in report.package_status] + report_dict['service_status'] = [asdict(s) for s in report.service_status] + + with open(filepath, 'w') as f: + json.dump(report_dict, f, indent=2) diff --git a/docs/SIMULATION_MODE.md b/docs/SIMULATION_MODE.md new file mode 100644 index 0000000..6f69e34 --- /dev/null +++ b/docs/SIMULATION_MODE.md @@ -0,0 +1,228 @@ +# Installation Simulation Mode Guide + +This guide covers the installation simulation feature for Cortex (Issue #103). + +## Overview + +The `--simulate` flag enables preview mode for installations, showing what would be installed without making any changes. This helps users: +- Preview what would be installed with **real package sizes from LLM** +- Check **actual disk space** availability on your system +- Verify system compatibility before installation +- See package information quickly + +### Key Features + +**Real System Checks:** +- āœ… Actual disk space detection (via `shutil.disk_usage()`) +- āœ… Real OS detection (platform module / `/etc/os-release`) +- āœ… Current package installation status +- āœ… Basic system information (kernel, architecture) + +**LLM-Powered Package Information:** +- šŸ¤– **Real package sizes** queried from OpenAI/Claude +- šŸ“¦ Accurate download sizes and version information +- šŸ”„ Auto-fallback: OpenAI ā†’ Anthropic +- šŸ“Š Estimates shown when no API key (marked with ~) + +## Usage + +### Basic Simulation + +```bash +cortex install docker --simulate +``` + +### Example Output (With API Key) + +``` +šŸ” Simulation mode: No changes will be made + +System Information: + OS: Windows 10.0.26200 + Kernel: 11 + Architecture: AMD64 + +Would install: + - containerd 1.4.3-1 (80 MB) + +Total download: 80 MB +Disk space required: 240 MB +Disk space available: 117 GB āœ“ + +Potential issues: None detected +``` + +### Example Output (Without API Key) + +``` +šŸ” Simulation mode: No changes will be made + +System Information: + OS: Ubuntu 22.04 + Kernel: 5.15.0 + Architecture: x86_64 + +Would install: + - docker-ce latest (~85 MB estimate) + - containerd.io latest (~45 MB estimate) + +Total download: ~130 MB (estimate) +Disk space required: ~390 MB (estimate) +Disk space available: 50 GB āœ“ + +Note: Install API key for accurate package sizes +Potential issues: None detected +``` + +### Simulating Different Software + +```bash +# Simulate Docker installation +cortex install docker --simulate + +# Simulate Python installation +cortex install python --simulate + +# Simulate nginx installation +cortex install nginx --simulate +``` + +## What Gets Checked + +### System Information +- Operating system (platform, version, distribution) +- Kernel version +- CPU architecture (amd64, arm64, x86_64) + +### Disk Space +- Available disk space on current directory +- Required space for installation (from LLM or estimates) + +### Package Status +- Docker installation check +- containerd installation check +- Generic software package detection + +### Package Information (via LLM) +- Real package sizes queried from OpenAI/Claude +- Package versions available +- Dependencies and download sizes +- Falls back to estimates if no API key + +## Report Information + +### Errors +Critical issues detected during simulation: +``` +āŒ Errors: + - Insufficient disk space: 200 MB available, 500 MB required + - Unable to detect OS information +``` + +### Warnings +Issues to be aware of (non-blocking): +``` +āš ļø Warnings: + - API key not found, using estimates for package sizes + - Package version could not be determined +``` + +## Combining with Other Flags + +The `--simulate` flag takes precedence: + +```bash +# This will only simulate, not execute +cortex install docker --simulate --execute + +# This will only simulate +cortex install docker --simulate --dry-run +``` + +## Exit Codes + +- `0`: Simulation completed, no blocking issues +- `1`: Simulation found blocking issues + +## Differences from --dry-run + +| Feature | --simulate | --dry-run | +|---------|-----------|-----------| +| System checks | Yes | No | +| API key required | Optional (better with) | Yes | +| Shows packages to install | Yes | No | +| Shows commands to run | No | Yes | +| Disk space analysis | Yes | No | +| LLM for package info | Yes | No | +| Package size accuracy | Real (with API key) | N/A | + +## Files Created + +- `cortex/preflight_checker.py` - Core preflight checking logic +- `test/test_preflight_checker.py` - Unit tests + +## LLM Integration + +### API Key Setup + +The simulation mode works best with an API key to query real package information: + +```bash +# Set OpenAI API key +export OPENAI_API_KEY="sk-..." + +# Or Anthropic API key (auto-fallback) +export ANTHROPIC_API_KEY="sk-ant-..." +``` + +Without an API key, the simulation uses estimated package sizes (marked with ~). + +### How It Works + +1. Detects your OS and system information +2. Queries LLM (OpenAI/Claude) for real package sizes +3. Auto-falls back to Anthropic if OpenAI fails +4. Uses estimates if no API key is available +5. Shows disk space requirements with āœ“ or āœ— + +## API Reference + +### PreflightChecker Class + +```python +from cortex.preflight_checker import PreflightChecker, format_report + +# Create checker with API key (optional) +checker = PreflightChecker(api_key="sk-...", provider="openai") + +# Run all checks for a package +report = checker.run_all_checks("docker") + +# Format and display report +output = format_report(report, "docker") +print(output) +``` + +### PreflightReport Fields + +- `os_info`: Operating system details (dict) +- `kernel_info`: Kernel version and architecture (dict) +- `disk_usage`: List of DiskInfo objects +- `package_status`: List of PackageInfo objects +- `packages_to_install`: List of packages to be installed +- `total_download_mb`: Total download size (float) +- `total_disk_required_mb`: Total disk space needed (float) +- `errors`: List of error messages +- `warnings`: List of warning messages + +### Helper Functions + +```python +# Export report to JSON +from cortex.preflight_checker import export_report +json_str = export_report(report) + +# Check specific software +pkg_info = checker.check_software("nginx") +print(f"{pkg_info.name}: installed={pkg_info.installed}, version={pkg_info.version}") +``` diff --git a/requirements.txt b/requirements.txt index 25a4cd2..8c44d95 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ openai>=1.0.0 # Type hints for older Python versions typing-extensions>=4.0.0 +PyYAML \ No newline at end of file diff --git a/test/test_cli.py b/test/test_cli.py index 635ad06..ec2b4ae 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -178,7 +178,7 @@ def test_main_install_command(self, mock_install): mock_install.return_value = 0 result = main() self.assertEqual(result, 0) - mock_install.assert_called_once_with('docker', execute=False, dry_run=False) + mock_install.assert_called_once_with('docker', execute=False, dry_run=False, simulate=False) @patch('sys.argv', ['cortex', 'install', 'docker', '--execute']) @patch('cortex.cli.CortexCLI.install') @@ -186,7 +186,7 @@ def test_main_install_with_execute(self, mock_install): mock_install.return_value = 0 result = main() self.assertEqual(result, 0) - mock_install.assert_called_once_with('docker', execute=True, dry_run=False) + mock_install.assert_called_once_with('docker', execute=True, dry_run=False, simulate=False) @patch('sys.argv', ['cortex', 'install', 'docker', '--dry-run']) @patch('cortex.cli.CortexCLI.install') @@ -194,7 +194,7 @@ def test_main_install_with_dry_run(self, mock_install): mock_install.return_value = 0 result = main() self.assertEqual(result, 0) - mock_install.assert_called_once_with('docker', execute=False, dry_run=True) + mock_install.assert_called_once_with('docker', execute=False, dry_run=True, simulate=False) def test_spinner_animation(self): initial_idx = self.cli.spinner_idx diff --git a/test/test_preflight_checker.py b/test/test_preflight_checker.py new file mode 100644 index 0000000..7a1b707 --- /dev/null +++ b/test/test_preflight_checker.py @@ -0,0 +1,201 @@ +""" +Unit tests for PreflightChecker + +Tests the preflight system checking functionality for installation simulation. +""" + +import unittest +from unittest.mock import patch, MagicMock +import sys +import os +import platform + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from cortex.preflight_checker import ( + PreflightChecker, + PreflightReport, + DiskInfo, + PackageInfo, + ServiceInfo, + format_report, + export_report +) + + +class TestPreflightChecker(unittest.TestCase): + + def setUp(self): + self.checker = PreflightChecker() # No API key for basic tests + + def test_check_os_info(self): + """Test OS information detection""" + info = self.checker.check_os_info() + + self.assertIn('platform', info) + self.assertIn('platform_release', info) + self.assertIn('machine', info) + self.assertIsNotNone(info['platform']) + + def test_check_basic_system_info(self): + """Test basic system information detection""" + info = self.checker.check_basic_system_info() + + self.assertIn('kernel', info) + self.assertIn('architecture', info) + self.assertIsNotNone(info['kernel']) + self.assertIsNotNone(info['architecture']) + + def test_check_disk_space(self): + """Test disk space checking""" + disk_info = self.checker.check_disk_space() + + self.assertIsInstance(disk_info, list) + self.assertTrue(len(disk_info) > 0) + + # Check that current directory is included + cwd = os.getcwd() + cwd_checked = any(d.path == cwd or cwd.startswith(d.path) for d in disk_info) + self.assertTrue(cwd_checked or len(disk_info) > 0) + + def test_check_software(self): + """Test software package detection""" + # Test with a common package + pkg = self.checker.check_software("curl") + + self.assertIsInstance(pkg, PackageInfo) + self.assertEqual(pkg.name, 'curl') + self.assertIsInstance(pkg.installed, bool) + + def test_run_all_checks(self): + """Test running all preflight checks""" + report = self.checker.run_all_checks("docker") + + self.assertIsInstance(report, PreflightReport) + self.assertIsNotNone(report.os_info) + self.assertIsNotNone(report.kernel_info) # Now contains basic system info + self.assertIsInstance(report.disk_usage, list) + self.assertIsInstance(report.package_status, list) + self.assertIsInstance(report.errors, list) + self.assertIsInstance(report.warnings, list) + + +class TestPreflightReport(unittest.TestCase): + + def test_report_dataclass(self): + """Test PreflightReport dataclass initialization""" + report = PreflightReport() + + self.assertEqual(report.os_info, {}) + self.assertEqual(report.kernel_info, {}) + self.assertEqual(report.cpu_arch, "") + self.assertEqual(report.disk_usage, []) + self.assertEqual(report.errors, []) + self.assertEqual(report.warnings, []) + self.assertEqual(report.suggestions, []) + + def test_disk_info_dataclass(self): + """Test DiskInfo dataclass""" + disk = DiskInfo( + path="/", + free_mb=50000, + total_mb=100000, + filesystem="ext4", + exists=True, + writable=True + ) + + self.assertEqual(disk.path, "/") + self.assertEqual(disk.free_mb, 50000) + self.assertTrue(disk.exists) + + def test_package_info_dataclass(self): + """Test PackageInfo dataclass""" + pkg = PackageInfo( + name="docker", + installed=True, + version="24.0.7", + path="/usr/bin/docker" + ) + + self.assertEqual(pkg.name, "docker") + self.assertTrue(pkg.installed) + self.assertEqual(pkg.version, "24.0.7") + + +class TestFormatReport(unittest.TestCase): + + def test_format_report_basic(self): + """Test report formatting""" + report = PreflightReport() + report.os_info = {'distro': 'Ubuntu', 'distro_version': '22.04'} + report.kernel_info = {'version': '5.15.0'} + report.cpu_arch = 'amd64' + report.packages_to_install = [ + {'name': 'docker-ce', 'version': 'latest', 'size_mb': '85'} + ] + report.total_download_mb = 85 + report.total_disk_required_mb = 255 + report.disk_usage = [ + DiskInfo(path='/', free_mb=50000, total_mb=100000, + filesystem='ext4', exists=True, writable=True) + ] + + output = format_report(report, "docker") + + self.assertIn("Simulation mode", output) + self.assertIn("docker-ce", output) + self.assertIn("85 MB", output) + + def test_format_report_no_install_needed(self): + """Test report when software is already installed""" + report = PreflightReport() + report.os_info = {'distro': 'Ubuntu', 'distro_version': '22.04'} + report.kernel_info = {'version': '5.15.0'} + report.cpu_arch = 'amd64' + report.packages_to_install = [] + report.disk_usage = [] + + output = format_report(report, "docker") + + self.assertIn("already installed", output) + + +class TestCLISimulateIntegration(unittest.TestCase): + + @patch('cortex.cli.PreflightChecker') + def test_install_simulate_flag(self, mock_checker_class): + """Test --simulate flag integration""" + from cortex.cli import CortexCLI + + mock_checker = MagicMock() + mock_report = PreflightReport() + mock_report.os_info = {'distro': 'Ubuntu', 'distro_version': '22.04'} + mock_report.kernel_info = {'version': '5.15.0'} + mock_report.cpu_arch = 'amd64' + mock_report.disk_usage = [] + mock_report.packages_to_install = [] + mock_checker.run_all_checks.return_value = mock_report + mock_checker_class.return_value = mock_checker + + cli = CortexCLI() + result = cli.install("docker", simulate=True) + + self.assertEqual(result, 0) + mock_checker.run_all_checks.assert_called_once_with("docker") + + @patch('sys.argv', ['cortex', 'install', 'docker', '--simulate']) + @patch('cortex.cli.CortexCLI.install') + def test_main_simulate_arg(self, mock_install): + """Test main function parses --simulate""" + from cortex.cli import main + + mock_install.return_value = 0 + result = main() + + self.assertEqual(result, 0) + mock_install.assert_called_once_with('docker', execute=False, dry_run=False, simulate=True) + + +if __name__ == '__main__': + unittest.main() From 91d4b9bc9e80339e5efd168a0d4011b6634ef45b Mon Sep 17 00:00:00 2001 From: sahil Date: Sun, 7 Dec 2025 18:18:39 +0530 Subject: [PATCH 2/3] Installation simulation mode fix #103 --- cortex/cli.py | 4 ++-- cortex/preflight_checker.py | 15 +++++---------- 2 files changed, 7 insertions(+), 12 deletions(-) diff --git a/cortex/cli.py b/cortex/cli.py index 1a4d6bd..13b8748 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -196,8 +196,8 @@ def progress_callback(current, total, step): def _run_simulation(self, software: str) -> int: """Run preflight simulation check for installation""" try: - # Get API key for LLM-powered package info (optional) - api_key = self._get_api_key() + # Get API key for LLM-powered package info (optional). + api_key = os.environ.get('OPENAI_API_KEY') or os.environ.get('ANTHROPIC_API_KEY') provider = self._get_provider() if api_key else 'openai' # Create checker with optional API key for enhanced accuracy diff --git a/cortex/preflight_checker.py b/cortex/preflight_checker.py index e1f8c4c..ce4f622 100644 --- a/cortex/preflight_checker.py +++ b/cortex/preflight_checker.py @@ -318,7 +318,6 @@ def _get_package_info_from_llm(self, software: str, os_info: Dict[str, str]) -> try: # Create a specific prompt for package information - distro = os_info.get('distro', 'Ubuntu') distro_id = os_info.get('distro_id', 'ubuntu') prompt = f"""Get {software} package information for Linux {distro_id}. @@ -466,7 +465,6 @@ def check_software(self, software_name: str) -> PackageInfo: def calculate_requirements(self, software: str) -> None: """Calculate installation requirements based on software to install""" - software_lower = software.lower() # Calculate total download and disk requirements total_download = 0 @@ -526,7 +524,7 @@ def run_all_checks(self, software: str = "docker") -> PreflightReport: def format_report(report: PreflightReport, software: str) -> str: """Format the preflight report for display""" lines = [] - lines.append(f"\nšŸ” Simulation mode: No changes will be made\n") + lines.append("\nšŸ” Simulation mode: No changes will be made\n") # Check if using estimates using_estimates = any('estimate' in str(pkg.get('size_mb', '')) for pkg in report.packages_to_install) @@ -541,7 +539,7 @@ def format_report(report: PreflightReport, software: str) -> str: # What would be installed if report.packages_to_install: - lines.append(f"\nWould install:") + lines.append("\nWould install:") for pkg in report.packages_to_install: lines.append(f" - {pkg['name']} {pkg.get('version', '')} ({pkg.get('size_mb', '?')} MB)") @@ -595,11 +593,8 @@ def export_report(report: PreflightReport, filepath: str) -> None: # Convert dataclass to dict report_dict = asdict(report) - - # Convert DiskInfo, PackageInfo, ServiceInfo to dicts - report_dict['disk_usage'] = [asdict(d) for d in report.disk_usage] - report_dict['package_status'] = [asdict(p) for p in report.package_status] - report_dict['service_status'] = [asdict(s) for s in report.service_status] - + + # `asdict` already converts nested dataclasses recursively, so we can + # directly write the result to JSON. with open(filepath, 'w') as f: json.dump(report_dict, f, indent=2) From 0ffa3d89b094f2acaf6632297d2c595f020589c2 Mon Sep 17 00:00:00 2001 From: sahil Date: Sun, 7 Dec 2025 18:32:06 +0530 Subject: [PATCH 3/3] Installation simulation mode fix #103 --- cortex/preflight_checker.py | 42 ++++++++++++++++++++++++++++++------- docs/SIMULATION_MODE.md | 2 +- 2 files changed, 35 insertions(+), 9 deletions(-) diff --git a/cortex/preflight_checker.py b/cortex/preflight_checker.py index ce4f622..4365bf6 100644 --- a/cortex/preflight_checker.py +++ b/cortex/preflight_checker.py @@ -76,8 +76,8 @@ class PreflightChecker: All checks are performed against the actual system - no simulation. """ - # Paths to check for disk space - DISK_CHECK_PATHS = ['/', '/var/lib/docker', '/opt'] + # Default paths - will be set based on platform in __init__ + DISK_CHECK_PATHS = [] MIN_DISK_SPACE_MB = 500 def __init__(self, api_key: Optional[str] = None, provider: str = 'openai'): @@ -88,6 +88,18 @@ def __init__(self, api_key: Optional[str] = None, provider: str = 'openai'): self.api_key = api_key self.provider = provider self._llm_client = None + + # Set platform-appropriate disk check paths + if self._is_windows: + # On Windows, check system drive and common program directories + system_drive = os.environ.get('SystemDrive', 'C:') + '\\' + self.DISK_CHECK_PATHS = [system_drive, os.path.join(system_drive, 'Program Files')] + elif self._is_mac: + # On macOS, check root and common application directories + self.DISK_CHECK_PATHS = ['/', '/Applications', '/usr/local'] + else: + # On Linux, check root and common Docker/application directories + self.DISK_CHECK_PATHS = ['/', '/var/lib/docker', '/opt'] def _run_command(self, cmd: List[str], timeout: int = 10) -> Tuple[bool, str]: """Run a shell command and return success status and output""" @@ -166,6 +178,7 @@ def _get_linux_distro(self) -> Dict[str, str]: elif key == 'ID': info['distro_id'] = value except Exception: + # If /etc/os-release is malformed or unreadable, continue with defaults pass return info @@ -238,6 +251,7 @@ def _get_disk_info(self, path: str) -> DiskInfo: free_mb = stat.free // (1024 * 1024) total_mb = stat.total // (1024 * 1024) except Exception: + # Path may not be accessible or permission denied; leave as 0 pass # Get filesystem type on Linux @@ -306,6 +320,7 @@ def _get_llm_client(self): self._llm_client = CommandInterpreter(api_key=fallback_key, provider=fallback_provider) return self._llm_client except Exception: + # Fallback provider also failed; will return None and use estimates pass return None @@ -365,8 +380,9 @@ def _get_package_info_from_llm(self, software: str, os_info: Dict[str, str]) -> data = json.loads(content) return data - except Exception: + except Exception as e: # LLM response parsing failed, fall back to estimates + self.report.warnings.append(f"Failed to parse LLM response for package info: {e}. Falling back to estimated package sizes.") pass return {'packages': [], 'total_size_mb': 0, 'config_changes': []} @@ -465,7 +481,6 @@ def check_software(self, software_name: str) -> PackageInfo: def calculate_requirements(self, software: str) -> None: """Calculate installation requirements based on software to install""" - # Calculate total download and disk requirements total_download = 0 total_disk = 0 @@ -479,14 +494,19 @@ def calculate_requirements(self, software: str) -> None: total_download += size total_disk += size * 3 # Rough estimate: downloaded + extracted + working space except (ValueError, AttributeError): - pass + # Could not parse size_mb for this package; skip and warn + self.report.warnings.append( + f"Could not parse size for package '{pkg.get('name', 'unknown')}', value: '{pkg.get('size_mb', '0')}'" + ) self.report.total_download_mb = total_download self.report.total_disk_required_mb = total_disk # Check if we have enough disk space + # Determine the root path for the current working directory + root_path = Path(os.getcwd()).anchor root_disk = next( - (d for d in self.report.disk_usage if d.path == '/'), + (d for d in self.report.disk_usage if os.path.normcase(os.path.abspath(d.path)) == os.path.normcase(os.path.abspath(root_path))), None ) @@ -554,7 +574,14 @@ def format_report(report: PreflightReport, software: str) -> str: lines.append(f"\nāœ“ {software} is already installed") # Disk space available - root_disk = next((d for d in report.disk_usage if d.path == '/'), None) + # Determine the root disk path based on the platform + if os.name == 'nt': + # On Windows, use the system drive (e.g., 'C:\\') + root_path = os.environ.get('SystemDrive', 'C:') + '\\' + else: + # On Unix-like systems, use '/' + root_path = '/' + root_disk = next((d for d in report.disk_usage if os.path.normcase(os.path.abspath(d.path)) == os.path.normcase(os.path.abspath(root_path))), None) if root_disk: status = 'āœ“' if root_disk.free_mb > report.total_disk_required_mb else 'āœ—' lines.append(f"Disk space available: {root_disk.free_mb // 1024} GB {status}") @@ -588,7 +615,6 @@ def format_report(report: PreflightReport, software: str) -> str: def export_report(report: PreflightReport, filepath: str) -> None: """Export preflight report to a JSON file""" - import json from dataclasses import asdict # Convert dataclass to dict diff --git a/docs/SIMULATION_MODE.md b/docs/SIMULATION_MODE.md index 6f69e34..5eaf1b2 100644 --- a/docs/SIMULATION_MODE.md +++ b/docs/SIMULATION_MODE.md @@ -220,7 +220,7 @@ print(output) ```python # Export report to JSON from cortex.preflight_checker import export_report -json_str = export_report(report) +export_report(report, "report.json") # Check specific software pkg_info = checker.check_software("nginx")