From d720aa605fedd62922898e87649b2ece5da512ea Mon Sep 17 00:00:00 2001 From: Krish Date: Thu, 18 Dec 2025 04:57:04 +0000 Subject: [PATCH 1/2] added doctor.py, pytest file, updated cli.py --- cortex/cli.py | 13 ++ cortex/doctor.py | 463 +++++++++++++++++++++++++++++++++++++++++++ tests/test_doctor.py | 133 +++++++++++++ 3 files changed, 609 insertions(+) create mode 100644 cortex/doctor.py create mode 100644 tests/test_doctor.py diff --git a/cortex/cli.py b/cortex/cli.py index e575db73..3a8ba5a4 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -171,6 +171,13 @@ def notify(self, args): # ------------------------------- + # Run system health checks + def doctor(self): + from cortex.doctor import SystemDoctor + + doctor = SystemDoctor() + return doctor.run_checks() + def install(self, software: str, execute: bool = False, dry_run: bool = False): # Validate input first is_valid, error = validate_install_request(software) @@ -562,6 +569,7 @@ def show_rich_help(): table.add_row("rollback ", "Undo installation") table.add_row("notify", "Manage desktop notifications") # Added this line table.add_row("cache stats", "Show LLM cache statistics") + table.add_row("doctor", "System health check") console.print(table) console.print() @@ -609,6 +617,9 @@ def main(): # Status command status_parser = subparsers.add_parser("status", help="Show system status") + # doctor command + doctor_parser = subparsers.add_parser("doctor", help="Run system health check") + # Install command install_parser = subparsers.add_parser("install", help="Install software") install_parser.add_argument("software", type=str, help="Software to install") @@ -688,6 +699,8 @@ def main(): # Handle the new notify command elif args.command == "notify": return cli.notify(args) + elif args.command == "doctor": + return cli.doctor() elif args.command == "cache": if getattr(args, "cache_action", None) == "stats": return cli.cache_stats() diff --git a/cortex/doctor.py b/cortex/doctor.py new file mode 100644 index 00000000..a6b53765 --- /dev/null +++ b/cortex/doctor.py @@ -0,0 +1,463 @@ +""" +System Health Check for Cortex Linux +Performs diagnostic checks and provides fix suggestions. +""" + +import os +import shutil +import subprocess +import sys +from pathlib import Path + +from rich import box +from rich.panel import Panel +from rich.table import Table + +from cortex.branding import console +from cortex.validators import validate_api_key + + +class SystemDoctor: + """ + Performs comprehensive system health checks and diagnostics. + + Checks for: + - Python version compatibility + - Required Python dependencies + - GPU drivers (NVIDIA/AMD) + - CUDA/ROCm availability + - Ollama installation and status + - API key configuration + - Disk space availability + - System memory + + Attributes: + warnings: List of non-critical issues found + failures: List of critical issues that may prevent operation + suggestions: List of fix commands for issues + passes: List of successful checks + """ + + def __init__(self) -> None: + self.warnings: list[str] = [] + self.failures: list[str] = [] + self.suggestions: list[str] = [] + self.passes: list[str] = [] + + def run_checks(self) -> int: + """ + Run all health checks and return appropriate exit code. + + Exit codes: + 0: All checks passed, system is healthy + 1: Warnings found, system can operate but has recommendations + 2: Critical failures found, system may not work properly + + Returns: + int: Exit code reflecting system health status (0, 1, or 2) + """ + # Header + console.print() + console.print( + Panel.fit( + "[bold cyan]CORTEX SYSTEM CHECK[/bold cyan]", border_style="cyan", padding=(1, 4) + ) + ) + console.print() + + # Run all check groups + self._print_section("Python & Dependencies") + self._check_python() + self._check_dependencies() + + self._print_section("GPU & Acceleration") + self._check_gpu_driver() + self._check_cuda() + + self._print_section("AI & Services") + self._check_ollama() + self._check_api_keys() + + self._print_section("System Resources") + self._check_disk_space() + self._check_memory() + + # Print summary + self._print_summary() + + # Return appropriate exit code + if self.failures: + return 2 # Critical failures + elif self.warnings: + return 1 # Warnings only + return 0 # All good + + def _print_section(self, title: str) -> None: + """Print a section header for grouping checks.""" + console.print(f"\n[bold cyan]{title}[/bold cyan]") + + def _print_check( + self, + status: str, + message: str, + suggestion: str | None = None, + ) -> None: + """ + Print a check result with appropriate formatting and colors. + + Args: + status: One of "PASS", "WARN", "FAIL", or "INFO" + message: Description of the check result + suggestion: Optional fix command or suggestion + """ + # Define symbols and colors + if status == "PASS": + symbol = "✓" + color = "bold green" + prefix = "[PASS]" + self.passes.append(message) + elif status == "WARN": + symbol = "⚠" + color = "bold yellow" + prefix = "[WARN]" + self.warnings.append(message) + if suggestion: + self.suggestions.append(suggestion) + elif status == "FAIL": + symbol = "✗" + color = "bold red" + prefix = "[FAIL]" + self.failures.append(message) + if suggestion: + self.suggestions.append(suggestion) + else: + symbol = "?" + color = "dim" + prefix = "[INFO]" + + # Print with icon prefix and coloring + console.print(f" [cyan]CX[/cyan] [{color}]{symbol} {prefix}[/{color}] {message}") + + def _check_python(self) -> None: + """Check Python version compatibility.""" + version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" + + if sys.version_info >= (3, 10): + self._print_check("PASS", f"Python {version}") + else: + self._print_check( + "FAIL", + f"Python {version} (3.10+ required)", + "Install Python 3.10+: sudo apt install python3.11", + ) + + def _check_dependencies(self) -> None: + """Check packages from requirements.txt.""" + missing: list[str] = [] + requirements_path = Path("requirements.txt") + + if not requirements_path.exists(): + self._print_check("WARN", "No requirements.txt found") + return + + # Map requirement names to importable module names + name_overrides = { + "pyyaml": "yaml", + "typing-extensions": "typing_extensions", + } + + try: + with open(requirements_path) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#"): + raw_name = line.split("==")[0].split(">")[0].split("<")[0] + pkg_name = name_overrides.get(raw_name, raw_name) + try: + __import__(pkg_name) + except ImportError: + missing.append(raw_name) + except Exception: + self._print_check("WARN", "Could not read requirements.txt") + return + + if not missing: + self._print_check("PASS", "All requirements.txt packages installed") + elif len(missing) < 3: + self._print_check( + "WARN", + f"Missing from requirements.txt: {', '.join(missing)}", + "Install dependencies: pip install -r requirements.txt", + ) + else: + self._print_check( + "FAIL", + f"Missing {len(missing)} packages from requirements.txt: {', '.join(missing[:3])}...", + "Install dependencies: pip install -r requirements.txt", + ) + + def _check_gpu_driver(self) -> None: + """Check for GPU drivers (NVIDIA or AMD ROCm).""" + # Check NVIDIA + if shutil.which("nvidia-smi"): + try: + result = subprocess.run( + ["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0 and result.stdout.strip(): + version = result.stdout.strip().split("\n")[0] + self._print_check("PASS", f"NVIDIA Driver {version}") + return + except (subprocess.TimeoutExpired, Exception): + pass + + # Check AMD ROCm + if shutil.which("rocm-smi"): + try: + result = subprocess.run( + ["rocm-smi", "--showdriverversion"], capture_output=True, text=True, timeout=5 + ) + if result.returncode == 0: + self._print_check("PASS", "AMD ROCm driver detected") + return + except (subprocess.TimeoutExpired, Exception): + pass + + # No GPU found - this is a warning, not a failure + self._print_check( + "WARN", + "No GPU detected (CPU-only mode supported, local inference will be slower)", # ← NEW + "Optional: Install NVIDIA/AMD drivers for acceleration", + ) + + def _check_cuda(self) -> None: + """Check CUDA/ROCm availability for GPU acceleration.""" + # Check CUDA + if shutil.which("nvcc"): + try: + result = subprocess.run( + ["nvcc", "--version"], capture_output=True, text=True, timeout=5 + ) + if result.returncode == 0 and "release" in result.stdout: + version_line = result.stdout.split("release")[1].split(",")[0].strip() + self._print_check("PASS", f"CUDA {version_line}") + return + except (subprocess.TimeoutExpired, Exception): + pass + + # Check ROCm + rocm_info_path = Path("/opt/rocm/.info/version") + if rocm_info_path.exists(): + try: + version = rocm_info_path.read_text(encoding="utf-8").strip() + self._print_check("PASS", f"ROCm {version}") + return + except (OSError, UnicodeDecodeError): + self._print_check("PASS", "ROCm installed") + return + elif Path("/opt/rocm").exists(): + self._print_check("PASS", "ROCm installed") + return + + # Check if PyTorch has CUDA available (software level) + try: + import torch + + if torch.cuda.is_available(): + self._print_check("PASS", "CUDA available (PyTorch)") + return + except ImportError: + pass + + self._print_check( + "WARN", + "CUDA/ROCm not found (GPU acceleration unavailable)", + "Install CUDA: https://developer.nvidia.com/cuda-downloads", + ) + + def _check_ollama(self) -> None: + """Check if Ollama is installed and running.""" + # Check if installed + if not shutil.which("ollama"): + self._print_check( + "WARN", + "Ollama not installed", + "Install Ollama: curl https://ollama.ai/install.sh | sh", + ) + return + + # Check if running by testing the API + try: + import requests + + response = requests.get("http://localhost:11434/api/tags", timeout=2) + if response.status_code == 200: + self._print_check("PASS", "Ollama installed and running") + return + except Exception: + pass + + # Ollama installed but not running + self._print_check( + "WARN", "Ollama installed but not running", "Start Ollama: ollama serve &" + ) + + def _check_api_keys(self) -> None: + """Check if API keys are configured for cloud models.""" + is_valid, provider, error = validate_api_key() + + if is_valid: + self._print_check("PASS", f"{provider} API key configured") + else: + self._print_check( + "WARN", + "No API keys configured (required for cloud models)", + "Configure API key: export ANTHROPIC_API_KEY=sk-... " "or run 'cortex wizard'", + ) + + def _check_disk_space(self) -> None: + """Check available disk space for model storage.""" + try: + usage = shutil.disk_usage(os.path.expanduser("~")) + free_gb = usage.free / (1024**3) + total_gb = usage.total / (1024**3) + + if free_gb > 20: + self._print_check( + "PASS", f"{free_gb:.1f}GB free disk space ({total_gb:.1f}GB total)" + ) + elif free_gb > 10: + self._print_check( + "WARN", + f"{free_gb:.1f}GB free (20GB+ recommended for models)", + "Free up disk space: sudo apt clean && docker system prune -a", + ) + else: + self._print_check( + "FAIL", + f"Only {free_gb:.1f}GB free (critically low)", + "Free up disk space: sudo apt autoremove && sudo apt clean", + ) + except (OSError, Exception) as e: + self._print_check("WARN", f"Could not check disk space: {type(e).__name__}") + + def _check_memory(self) -> None: + """Check system RAM availability.""" + mem_gb = self._get_system_memory() + + if mem_gb is None: + self._print_check("WARN", "Could not detect system RAM") + return + + if mem_gb >= 16: + self._print_check("PASS", f"{mem_gb:.1f}GB RAM") + elif mem_gb >= 8: + self._print_check( + "WARN", + f"{mem_gb:.1f}GB RAM (16GB recommended for larger models)", + "Consider upgrading RAM or use smaller models", + ) + else: + self._print_check( + "FAIL", + f"Only {mem_gb:.1f}GB RAM (8GB minimum required)", + "Upgrade RAM to at least 8GB", + ) + + def _get_system_memory(self) -> float | None: + """ + Get system memory in GB. + + Returns: + float: Total system memory in GB, or None if detection fails + """ + # Try /proc/meminfo (Linux) + try: + with open("/proc/meminfo", encoding="utf-8") as f: + for line in f: + if line.startswith("MemTotal:"): + mem_kb = int(line.split()[1]) + return mem_kb / (1024**2) + except (OSError, ValueError, IndexError): + pass + + # Try psutil (macOS/BSD/Windows) + try: + import psutil + + return psutil.virtual_memory().total / (1024**3) + except ImportError: + pass + + return None + + def _print_summary(self) -> None: + """Print summary table and overall health status with suggestions.""" + console.print() + + # Create summary table + table = Table(show_header=False, box=box.SIMPLE, padding=(0, 1)) + table.add_column("Status", style="bold") + table.add_column("Count", justify="right") + + if self.passes: + table.add_row("[green]✓ Passed[/green]", f"[green]{len(self.passes)}[/green]") + if self.warnings: + table.add_row("[yellow]⚠ Warnings[/yellow]", f"[yellow]{len(self.warnings)}[/yellow]") + if self.failures: + table.add_row("[red]✗ Failures[/red]", f"[red]{len(self.failures)}[/red]") + + console.print(table) + console.print() + + # Overall status panel + if self.failures: + console.print( + Panel( + f"[bold red]❌ {len(self.failures)} critical failure(s) found[/bold red]", + border_style="red", + padding=(0, 2), + ) + ) + elif self.warnings: + console.print( + Panel( + f"[bold yellow]⚠️ {len(self.warnings)} warning(s) found[/bold yellow]", + border_style="yellow", + padding=(0, 2), + ) + ) + else: + console.print( + Panel( + "[bold green]✅ All checks passed! System is healthy.[/bold green]", + border_style="green", + padding=(0, 2), + ) + ) + + # Show fix suggestions if any + if self.suggestions: + console.print() + console.print("[bold cyan]💡 Suggested fixes:[/bold cyan]") + for i, suggestion in enumerate(self.suggestions, 1): + console.print(f" [dim]{i}.[/dim] {suggestion}") + console.print() + + +def run_doctor() -> int: + """ + Run the system doctor and return exit code. + + Returns: + int: Exit code (0 = all good, 1 = warnings, 2 = failures) + """ + doctor = SystemDoctor() + return doctor.run_checks() + + +if __name__ == "__main__": + sys.exit(run_doctor()) diff --git a/tests/test_doctor.py b/tests/test_doctor.py new file mode 100644 index 00000000..6cdb1ce4 --- /dev/null +++ b/tests/test_doctor.py @@ -0,0 +1,133 @@ +""" +Unit tests for cortex/doctor.py - System Health Check +""" + +import sys +from collections import namedtuple +from unittest.mock import MagicMock, mock_open, patch + +import pytest + +from cortex.doctor import SystemDoctor + + +class TestSystemDoctorInit: + def test_init_empty_lists(self): + doctor = SystemDoctor() + assert doctor.passes == [] + assert doctor.warnings == [] + assert doctor.failures == [] + assert doctor.suggestions == [] + + +class TestPythonVersionCheck: + VersionInfo = namedtuple("VersionInfo", "major minor micro releaselevel serial") + + @pytest.mark.parametrize( + "version_tuple, status", + [ + ((3, 12, 3), "PASS"), + ((3, 9, 0), "WARN"), + ((3, 7, 0), "FAIL"), + ], + ) + def test_python_version_scenarios(self, monkeypatch, version_tuple, status): + doctor = SystemDoctor() + + vi = self.VersionInfo(version_tuple[0], version_tuple[1], version_tuple[2], "final", 0) + monkeypatch.setattr(sys, "version_info", vi) + + doctor._check_python() + + if status == "PASS": + assert any("Python 3.12.3" in msg for msg in doctor.passes) + elif status == "WARN": + assert any("Python 3.9.0" in msg for msg in doctor.warnings) + else: + assert any("Python 3.7.0" in msg for msg in doctor.failures) + + +class TestRequirementsTxtDependencies: + def test_requirements_txt_all_installed(self): + doctor = SystemDoctor() + mock_content = "anthropic\nopenai\nrich\n" + + with patch("builtins.open", mock_open(read_data=mock_content)): + with patch("builtins.__import__", return_value=MagicMock()): + doctor._check_dependencies() + + # OLD (failing) + # assert "dependencies installed" in doctor.passes[0] + + # NEW (matches your doctor.py output) + assert "All requirements.txt packages installed" in doctor.passes[0] + + def test_some_dependencies_missing(self): + doctor = SystemDoctor() + + def mock_import(name, *args, **kwargs): + if name in ["anthropic", "openai"]: + raise ImportError() + return MagicMock() + + mock_content = "anthropic\nopenai\nrich\n" + with patch("builtins.open", mock_open(read_data=mock_content)): + with patch("builtins.__import__", side_effect=mock_import): + doctor._check_dependencies() + + # OLD (failing) + # assert "Missing dependencies" in doctor.warnings[0] + + # NEW (matches your doctor.py output) + assert "Missing from requirements.txt" in doctor.warnings[0] + + +class TestGPUDriverCheck: + def test_cpu_only_message(self): + doctor = SystemDoctor() + with patch("shutil.which", return_value=None): + doctor._check_gpu_driver() + assert "CPU-only mode" in doctor.warnings[0] + + +class TestExitCodes: + """ + IMPORTANT: run_checks() calls all checks; without patching, your real system + will produce warnings/failures and exit code 2, which is why your previous + tests saw 2 instead of 1/0. + """ + + @patch.object(SystemDoctor, "_check_python") + @patch.object(SystemDoctor, "_check_dependencies") + @patch.object(SystemDoctor, "_check_gpu_driver") + @patch.object(SystemDoctor, "_check_cuda") + @patch.object(SystemDoctor, "_check_ollama") + @patch.object(SystemDoctor, "_check_api_keys") + @patch.object(SystemDoctor, "_check_disk_space") + @patch.object(SystemDoctor, "_check_memory") + @patch.object(SystemDoctor, "_print_summary") + def test_exit_codes(self, *_mocks): + # all good → 0 + d = SystemDoctor() + d.passes = ["ok"] + d.warnings = [] + d.failures = [] + assert d.run_checks() == 0 + + # warnings only → 1 + d = SystemDoctor() + d.passes = ["ok"] + d.warnings = ["warn"] + d.failures = [] + assert d.run_checks() == 1 + + # failures present → 2 + d = SystemDoctor() + d.passes = ["ok"] + d.warnings = ["warn"] + d.failures = ["fail"] + assert d.run_checks() == 2 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) From 8cff105e9c0a62c74e7b55eab4b47bc5f83eb316 Mon Sep 17 00:00:00 2001 From: Krish Date: Thu, 18 Dec 2025 06:00:14 +0000 Subject: [PATCH 2/2] solved coderabbitai errors, ci erros and Cortex Automation / test errors --- cortex/doctor.py | 2 +- tests/test_doctor.py | 40 +++++++++++++++++----------------------- 2 files changed, 18 insertions(+), 24 deletions(-) diff --git a/cortex/doctor.py b/cortex/doctor.py index a6b53765..057bed87 100644 --- a/cortex/doctor.py +++ b/cortex/doctor.py @@ -142,7 +142,7 @@ def _check_python(self) -> None: """Check Python version compatibility.""" version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" - if sys.version_info >= (3, 10): + if sys.version_info >= (3, 10): # noqa: UP036 self._print_check("PASS", f"Python {version}") else: self._print_check( diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 6cdb1ce4..a073a42d 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -27,7 +27,7 @@ class TestPythonVersionCheck: "version_tuple, status", [ ((3, 12, 3), "PASS"), - ((3, 9, 0), "WARN"), + ((3, 9, 0), "FAIL"), ((3, 7, 0), "FAIL"), ], ) @@ -39,12 +39,12 @@ def test_python_version_scenarios(self, monkeypatch, version_tuple, status): doctor._check_python() + version_str = f"Python {version_tuple[0]}.{version_tuple[1]}.{version_tuple[2]}" + if status == "PASS": - assert any("Python 3.12.3" in msg for msg in doctor.passes) - elif status == "WARN": - assert any("Python 3.9.0" in msg for msg in doctor.warnings) + assert any(version_str in msg for msg in doctor.passes) else: - assert any("Python 3.7.0" in msg for msg in doctor.failures) + assert any(version_str in msg for msg in doctor.failures) class TestRequirementsTxtDependencies: @@ -52,34 +52,28 @@ def test_requirements_txt_all_installed(self): doctor = SystemDoctor() mock_content = "anthropic\nopenai\nrich\n" - with patch("builtins.open", mock_open(read_data=mock_content)): - with patch("builtins.__import__", return_value=MagicMock()): - doctor._check_dependencies() - - # OLD (failing) - # assert "dependencies installed" in doctor.passes[0] + with patch("pathlib.Path.exists", return_value=True): + with patch("builtins.open", mock_open(read_data=mock_content)): + with patch("builtins.__import__", return_value=MagicMock()): + doctor._check_dependencies() - # NEW (matches your doctor.py output) assert "All requirements.txt packages installed" in doctor.passes[0] def test_some_dependencies_missing(self): doctor = SystemDoctor() + mock_content = "anthropic\nopenai\nrich\n" - def mock_import(name, *args, **kwargs): - if name in ["anthropic", "openai"]: + def fake_import(name, *args, **kwargs): + if name == "openai": raise ImportError() return MagicMock() - mock_content = "anthropic\nopenai\nrich\n" - with patch("builtins.open", mock_open(read_data=mock_content)): - with patch("builtins.__import__", side_effect=mock_import): - doctor._check_dependencies() - - # OLD (failing) - # assert "Missing dependencies" in doctor.warnings[0] + with patch("pathlib.Path.exists", return_value=True): + with patch("builtins.open", mock_open(read_data=mock_content)): + with patch("builtins.__import__", side_effect=fake_import): + doctor._check_dependencies() - # NEW (matches your doctor.py output) - assert "Missing from requirements.txt" in doctor.warnings[0] + assert any("Missing from requirements.txt: openai" in msg for msg in doctor.warnings) class TestGPUDriverCheck: