diff --git a/.github/scripts/cla_check.py b/.github/scripts/cla_check.py index 64c72bc0..75a4d7de 100644 --- a/.github/scripts/cla_check.py +++ b/.github/scripts/cla_check.py @@ -8,6 +8,7 @@ import os import re import sys + import requests # Configuration @@ -85,11 +86,7 @@ def load_cla_signers() -> dict: sys.exit(1) -def is_signer( - username: str | None, - email: str, - signers: dict -) -> tuple[bool, str | None]: +def is_signer(username: str | None, email: str, signers: dict) -> tuple[bool, str | None]: """ Check if a user has signed the CLA. Returns (is_signed, signing_entity). @@ -129,12 +126,7 @@ def is_signer( return False, None -def get_pr_authors( - owner: str, - repo: str, - pr_number: int, - token: str -) -> list[dict]: +def get_pr_authors(owner: str, repo: str, pr_number: int, token: str) -> list[dict]: """ Get all unique authors from PR commits. Returns list of {username, email, name, source}. @@ -142,10 +134,7 @@ def get_pr_authors( authors = {} # Get PR commits - commits = github_request( - f"repos/{owner}/{repo}/pulls/{pr_number}/commits?per_page=100", - token - ) + commits = github_request(f"repos/{owner}/{repo}/pulls/{pr_number}/commits?per_page=100", token) for commit in commits: sha = commit["sha"] @@ -167,7 +156,7 @@ def get_pr_authors( "username": author_username, "email": author_email, "name": author_name, - "source": f"commit {sha[:7]}" + "source": f"commit {sha[:7]}", } # Committer (if different) @@ -185,7 +174,7 @@ def get_pr_authors( "username": committer_username, "email": committer_email, "name": committer_name, - "source": f"committer {sha[:7]}" + "source": f"committer {sha[:7]}", } # Co-authors from commit message @@ -197,7 +186,7 @@ def get_pr_authors( "username": None, "email": co_email, "name": co_name, - "source": f"co-author {sha[:7]}" + "source": f"co-author {sha[:7]}", } return list(authors.values()) @@ -209,7 +198,7 @@ def post_comment( pr_number: int, token: str, missing_authors: list[dict], - signed_authors: list[tuple[dict, str]] + signed_authors: list[tuple[dict, str]], ) -> None: """Post or update CLA status comment on PR.""" # Build comment body @@ -250,8 +239,7 @@ def post_comment( # Check for existing CLA comment to update comments = github_request( - f"repos/{owner}/{repo}/issues/{pr_number}/comments?per_page=100", - token + f"repos/{owner}/{repo}/issues/{pr_number}/comments?per_page=100", token ) cla_comment_id = None @@ -269,23 +257,17 @@ def post_comment( requests.patch( f"{GITHUB_API}/repos/{owner}/{repo}/issues/comments/{cla_comment_id}", headers=headers, - json={"body": comment_body} + json={"body": comment_body}, ) else: # Create new comment github_post( - f"repos/{owner}/{repo}/issues/{pr_number}/comments", - token, - {"body": comment_body} + f"repos/{owner}/{repo}/issues/{pr_number}/comments", token, {"body": comment_body} ) def post_success_comment( - owner: str, - repo: str, - pr_number: int, - token: str, - signed_authors: list[tuple[dict, str]] + owner: str, repo: str, pr_number: int, token: str, signed_authors: list[tuple[dict, str]] ) -> None: """Post success comment or update existing CLA comment.""" lines = ["## CLA Verification Passed\n\n"] @@ -306,8 +288,7 @@ def post_success_comment( # Check for existing CLA comment to update comments = github_request( - f"repos/{owner}/{repo}/issues/{pr_number}/comments?per_page=100", - token + f"repos/{owner}/{repo}/issues/{pr_number}/comments?per_page=100", token ) for comment in comments: @@ -320,7 +301,7 @@ def post_success_comment( requests.patch( f"{GITHUB_API}/repos/{owner}/{repo}/issues/comments/{comment['id']}", headers=headers, - json={"body": comment_body} + json={"body": comment_body}, ) return @@ -328,9 +309,7 @@ def post_success_comment( # (single author PRs don't need a "you signed" comment) if len(signed_authors) > 1: github_post( - f"repos/{owner}/{repo}/issues/{pr_number}/comments", - token, - {"body": comment_body} + f"repos/{owner}/{repo}/issues/{pr_number}/comments", token, {"body": comment_body} ) @@ -358,8 +337,14 @@ def main(): # Allowlist for bots bot_patterns = [ - "dependabot", "github-actions", "renovate", "codecov", - "sonarcloud", "coderabbitai", "sonarqubecloud", "noreply@github.com" + "dependabot", + "github-actions", + "renovate", + "codecov", + "sonarcloud", + "coderabbitai", + "sonarqubecloud", + "noreply@github.com", ] for author in authors: diff --git a/README.md b/README.md index 1e550b78..835fcf0f 100644 --- a/README.md +++ b/README.md @@ -145,6 +145,7 @@ cortex rollback | `cortex install ` | Install packages matching natural language query | | `cortex install --dry-run` | Preview installation plan (default) | | `cortex install --execute` | Execute the installation | +| `cortex sandbox ` | Test packages in Docker sandbox | | `cortex history` | View all past installations | | `cortex rollback ` | Undo a specific installation | | `cortex --version` | Show version information | diff --git a/cortex/cli.py b/cortex/cli.py index a0bf8d0f..7d248002 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -286,6 +286,226 @@ def _handle_stack_real_install(self, stack: dict[str, Any], packages: list[str]) console.print(f"Installed {len(packages)} packages") return 0 + # --- Sandbox Commands (Docker-based package testing) --- + def sandbox(self, args: argparse.Namespace) -> int: + """Handle `cortex sandbox` commands for Docker-based package testing.""" + from cortex.sandbox import ( + DockerNotFoundError, + DockerSandbox, + SandboxAlreadyExistsError, + SandboxNotFoundError, + SandboxTestStatus, + ) + + action = getattr(args, "sandbox_action", None) + + if not action: + cx_print("\nšŸ³ Docker Sandbox - Test packages safely before installing\n", "info") + console.print("Usage: cortex sandbox [options]") + console.print("\nCommands:") + console.print(" create Create a sandbox environment") + console.print(" install Install package in sandbox") + console.print(" test [package] Run tests in sandbox") + console.print(" promote Install tested package on main system") + console.print(" cleanup Remove sandbox environment") + console.print(" list List all sandboxes") + console.print(" exec Execute command in sandbox") + console.print("\nExample workflow:") + console.print(" cortex sandbox create test-env") + console.print(" cortex sandbox install test-env nginx") + console.print(" cortex sandbox test test-env") + console.print(" cortex sandbox promote test-env nginx") + console.print(" cortex sandbox cleanup test-env") + return 0 + + try: + sandbox = DockerSandbox() + + if action == "create": + return self._sandbox_create(sandbox, args) + elif action == "install": + return self._sandbox_install(sandbox, args) + elif action == "test": + return self._sandbox_test(sandbox, args) + elif action == "promote": + return self._sandbox_promote(sandbox, args) + elif action == "cleanup": + return self._sandbox_cleanup(sandbox, args) + elif action == "list": + return self._sandbox_list(sandbox) + elif action == "exec": + return self._sandbox_exec(sandbox, args) + else: + self._print_error(f"Unknown sandbox action: {action}") + return 1 + + except DockerNotFoundError as e: + self._print_error(str(e)) + cx_print("Docker is required only for sandbox commands.", "info") + return 1 + except SandboxNotFoundError as e: + self._print_error(str(e)) + cx_print("Use 'cortex sandbox list' to see available sandboxes.", "info") + return 1 + except SandboxAlreadyExistsError as e: + self._print_error(str(e)) + return 1 + + def _sandbox_create(self, sandbox, args: argparse.Namespace) -> int: + """Create a new sandbox environment.""" + name = args.name + image = getattr(args, "image", "ubuntu:22.04") + + cx_print(f"Creating sandbox '{name}'...", "info") + result = sandbox.create(name, image=image) + + if result.success: + cx_print(f"āœ“ Sandbox environment '{name}' created", "success") + console.print(f" [dim]{result.stdout}[/dim]") + return 0 + else: + self._print_error(result.message) + if result.stderr: + console.print(f" [red]{result.stderr}[/red]") + return 1 + + def _sandbox_install(self, sandbox, args: argparse.Namespace) -> int: + """Install a package in sandbox.""" + name = args.name + package = args.package + + cx_print(f"Installing '{package}' in sandbox '{name}'...", "info") + result = sandbox.install(name, package) + + if result.success: + cx_print(f"āœ“ {package} installed in sandbox", "success") + return 0 + else: + self._print_error(result.message) + if result.stderr: + console.print(f" [dim]{result.stderr[:500]}[/dim]") + return 1 + + def _sandbox_test(self, sandbox, args: argparse.Namespace) -> int: + """Run tests in sandbox.""" + from cortex.sandbox import SandboxTestStatus + + name = args.name + package = getattr(args, "package", None) + + cx_print(f"Running tests in sandbox '{name}'...", "info") + result = sandbox.test(name, package) + + console.print() + for test in result.test_results: + if test.result == SandboxTestStatus.PASSED: + console.print(f" āœ“ {test.name}") + if test.message: + console.print(f" [dim]{test.message[:80]}[/dim]") + elif test.result == SandboxTestStatus.FAILED: + console.print(f" āœ— {test.name}") + if test.message: + console.print(f" [red]{test.message}[/red]") + else: + console.print(f" āŠ˜ {test.name} [dim](skipped)[/dim]") + + console.print() + if result.success: + cx_print("All tests passed", "success") + return 0 + else: + self._print_error("Some tests failed") + return 1 + + def _sandbox_promote(self, sandbox, args: argparse.Namespace) -> int: + """Promote a tested package to main system.""" + name = args.name + package = args.package + dry_run = getattr(args, "dry_run", False) + skip_confirm = getattr(args, "yes", False) + + if dry_run: + result = sandbox.promote(name, package, dry_run=True) + cx_print(f"Would run: sudo apt-get install -y {package}", "info") + return 0 + + # Confirm with user unless -y flag + if not skip_confirm: + console.print(f"\nPromote '{package}' to main system? [Y/n]: ", end="") + try: + response = input().strip().lower() + if response and response not in ("y", "yes"): + cx_print("Promotion cancelled", "warning") + return 0 + except (EOFError, KeyboardInterrupt): + console.print() + cx_print("Promotion cancelled", "warning") + return 0 + + cx_print(f"Installing '{package}' on main system...", "info") + result = sandbox.promote(name, package, dry_run=False) + + if result.success: + cx_print(f"āœ“ {package} installed on main system", "success") + return 0 + else: + self._print_error(result.message) + if result.stderr: + console.print(f" [red]{result.stderr[:500]}[/red]") + return 1 + + def _sandbox_cleanup(self, sandbox, args: argparse.Namespace) -> int: + """Remove a sandbox environment.""" + name = args.name + force = getattr(args, "force", False) + + cx_print(f"Removing sandbox '{name}'...", "info") + result = sandbox.cleanup(name, force=force) + + if result.success: + cx_print(f"āœ“ Sandbox '{name}' removed", "success") + return 0 + else: + self._print_error(result.message) + return 1 + + def _sandbox_list(self, sandbox) -> int: + """List all sandbox environments.""" + sandboxes = sandbox.list_sandboxes() + + if not sandboxes: + cx_print("No sandbox environments found", "info") + cx_print("Create one with: cortex sandbox create ", "info") + return 0 + + cx_print("\nšŸ³ Sandbox Environments:\n", "info") + for sb in sandboxes: + status_icon = "šŸŸ¢" if sb.state.value == "running" else "āšŖ" + console.print(f" {status_icon} [green]{sb.name}[/green]") + console.print(f" Image: {sb.image}") + console.print(f" Created: {sb.created_at[:19]}") + if sb.packages: + console.print(f" Packages: {', '.join(sb.packages)}") + console.print() + + return 0 + + def _sandbox_exec(self, sandbox, args: argparse.Namespace) -> int: + """Execute command in sandbox.""" + name = args.name + command = args.command + + result = sandbox.exec_command(name, command) + + if result.stdout: + console.print(result.stdout, end="") + if result.stderr: + console.print(result.stderr, style="red", end="") + + return result.exit_code + + # --- End Sandbox Commands --- + def ask(self, question: str) -> int: """Answer a natural language question about the system.""" api_key = self._get_api_key() @@ -1332,6 +1552,7 @@ def show_rich_help(): table.add_row("env", "Manage environment variables") table.add_row("cache stats", "Show LLM cache statistics") table.add_row("stack ", "Install the stack") + table.add_row("sandbox ", "Test packages in Docker sandbox") table.add_row("doctor", "System health check") console.print(table) @@ -1496,6 +1717,56 @@ def main(): cache_subs = cache_parser.add_subparsers(dest="cache_action", help="Cache actions") cache_subs.add_parser("stats", help="Show cache statistics") + # --- Sandbox Commands (Docker-based package testing) --- + sandbox_parser = subparsers.add_parser( + "sandbox", help="Test packages in isolated Docker sandbox" + ) + sandbox_subs = sandbox_parser.add_subparsers(dest="sandbox_action", help="Sandbox actions") + + # sandbox create [--image IMAGE] + sandbox_create_parser = sandbox_subs.add_parser("create", help="Create a sandbox environment") + sandbox_create_parser.add_argument("name", help="Unique name for the sandbox") + sandbox_create_parser.add_argument( + "--image", default="ubuntu:22.04", help="Docker image to use (default: ubuntu:22.04)" + ) + + # sandbox install + sandbox_install_parser = sandbox_subs.add_parser("install", help="Install a package in sandbox") + sandbox_install_parser.add_argument("name", help="Sandbox name") + sandbox_install_parser.add_argument("package", help="Package to install") + + # sandbox test [package] + sandbox_test_parser = sandbox_subs.add_parser("test", help="Run tests in sandbox") + sandbox_test_parser.add_argument("name", help="Sandbox name") + sandbox_test_parser.add_argument("package", nargs="?", help="Specific package to test") + + # sandbox promote [--dry-run] + sandbox_promote_parser = sandbox_subs.add_parser( + "promote", help="Install tested package on main system" + ) + sandbox_promote_parser.add_argument("name", help="Sandbox name") + sandbox_promote_parser.add_argument("package", help="Package to promote") + sandbox_promote_parser.add_argument( + "--dry-run", action="store_true", help="Show command without executing" + ) + sandbox_promote_parser.add_argument( + "-y", "--yes", action="store_true", help="Skip confirmation prompt" + ) + + # sandbox cleanup [--force] + sandbox_cleanup_parser = sandbox_subs.add_parser("cleanup", help="Remove a sandbox environment") + sandbox_cleanup_parser.add_argument("name", help="Sandbox name to remove") + sandbox_cleanup_parser.add_argument("-f", "--force", action="store_true", help="Force removal") + + # sandbox list + sandbox_subs.add_parser("list", help="List all sandbox environments") + + # sandbox exec + sandbox_exec_parser = sandbox_subs.add_parser("exec", help="Execute command in sandbox") + sandbox_exec_parser.add_argument("name", help="Sandbox name") + sandbox_exec_parser.add_argument("command", nargs="+", help="Command to execute") + # -------------------------- + # --- Environment Variable Management Commands --- env_parser = subparsers.add_parser("env", help="Manage environment variables") env_subs = env_parser.add_subparsers(dest="env_action", help="Environment actions") @@ -1623,6 +1894,8 @@ def main(): return cli.notify(args) elif args.command == "stack": return cli.stack(args) + elif args.command == "sandbox": + return cli.sandbox(args) elif args.command == "cache": if getattr(args, "cache_action", None) == "stats": return cli.cache_stats() diff --git a/cortex/sandbox/__init__.py b/cortex/sandbox/__init__.py index e69de29b..6371f87f 100644 --- a/cortex/sandbox/__init__.py +++ b/cortex/sandbox/__init__.py @@ -0,0 +1,40 @@ +""" +Cortex Sandbox Module + +Provides sandboxed execution environments for safe package testing. + +- SandboxExecutor: Firejail-based command sandboxing +- DockerSandbox: Docker-based package testing environments +""" + +from cortex.sandbox.docker_sandbox import ( + DockerNotFoundError, + DockerSandbox, + SandboxAlreadyExistsError, + SandboxExecutionResult, + SandboxInfo, + SandboxNotFoundError, + SandboxState, + SandboxTestResult, + SandboxTestStatus, + docker_available, +) +from cortex.sandbox.sandbox_executor import CommandBlocked, ExecutionResult, SandboxExecutor + +__all__ = [ + # Firejail sandbox + "CommandBlocked", + "ExecutionResult", + "SandboxExecutor", + # Docker sandbox + "DockerNotFoundError", + "DockerSandbox", + "SandboxAlreadyExistsError", + "SandboxExecutionResult", + "SandboxInfo", + "SandboxNotFoundError", + "SandboxState", + "SandboxTestResult", + "SandboxTestStatus", + "docker_available", +] diff --git a/cortex/sandbox/docker_sandbox.py b/cortex/sandbox/docker_sandbox.py new file mode 100644 index 00000000..71e57fc8 --- /dev/null +++ b/cortex/sandbox/docker_sandbox.py @@ -0,0 +1,910 @@ +#!/usr/bin/env python3 +""" +Docker-based Package Sandbox Testing Environment for Cortex Linux. + +Provides isolated Docker containers for testing packages before installing +to the main system. Docker is required only for sandbox commands. + +Features: +- Create isolated Docker environments +- Install packages in sandbox +- Run automated tests +- Validate functionality +- Promote to main system (fresh install on host) +- Automatic cleanup +""" + +import json +import logging +import os +import shutil +import subprocess +import time +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + + +class SandboxState(Enum): + """State of a sandbox environment.""" + + CREATED = "created" + RUNNING = "running" + STOPPED = "stopped" + DESTROYED = "destroyed" + + +class SandboxTestStatus(Enum): + """Result of a sandbox test.""" + + PASSED = "passed" + FAILED = "failed" + SKIPPED = "skipped" + + +@dataclass +class SandboxTestResult: + """Result of a single test in sandbox.""" + + name: str + result: SandboxTestStatus + message: str = "" + duration: float = 0.0 + + +@dataclass +class SandboxInfo: + """Information about a sandbox environment.""" + + name: str + container_id: str + state: SandboxState + created_at: str + image: str + packages: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary.""" + return { + "name": self.name, + "container_id": self.container_id, + "state": self.state.value, + "created_at": self.created_at, + "image": self.image, + "packages": self.packages, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "SandboxInfo": + """Create from dictionary.""" + return cls( + name=data["name"], + container_id=data["container_id"], + state=SandboxState(data["state"]), + created_at=data["created_at"], + image=data["image"], + packages=data.get("packages", []), + ) + + +@dataclass +class SandboxExecutionResult: + """Result of sandbox operation.""" + + success: bool + message: str + exit_code: int = 0 + stdout: str = "" + stderr: str = "" + test_results: list[SandboxTestResult] = field(default_factory=list) + packages_installed: list[str] = field(default_factory=list) + + +class DockerNotFoundError(Exception): + """Raised when Docker is not installed or not running.""" + + pass + + +class SandboxNotFoundError(Exception): + """Raised when a sandbox environment is not found.""" + + pass + + +class SandboxAlreadyExistsError(Exception): + """Raised when trying to create a sandbox that already exists.""" + + pass + + +class DockerSandbox: + """ + Docker-based sandbox manager for package testing. + + Provides isolated environments using Docker containers for safe + package testing before installation on the host system. + + Example: + sandbox = DockerSandbox() + sandbox.create("test-env") + sandbox.install("test-env", "nginx") + result = sandbox.test("test-env") + if result.success: + sandbox.promote("test-env", "nginx") + sandbox.cleanup("test-env") + """ + + # Default base image for sandboxes + DEFAULT_IMAGE = "ubuntu:22.04" + + # Container name prefix + CONTAINER_PREFIX = "cortex-sandbox-" + + # Commands that cannot run in Docker sandbox + SANDBOX_BLOCKED_COMMANDS = { + "systemctl", + "service", + "journalctl", + "modprobe", + "insmod", + "rmmod", + "lsmod", + "sysctl", + "mount", + "umount", + "fdisk", + "mkfs", + "reboot", + "shutdown", + "halt", + "poweroff", + "init", + } + + def __init__( + self, + data_dir: Path | None = None, + image: str | None = None, + ): + """ + Initialize Docker sandbox manager. + + Args: + data_dir: Directory to store sandbox metadata. Defaults to ~/.cortex/sandboxes + image: Default Docker image to use. Defaults to ubuntu:22.04 + """ + self.data_dir = data_dir or Path.home() / ".cortex" / "sandboxes" + self.default_image = image or self.DEFAULT_IMAGE + self._docker_path: str | None = None + + # Ensure data directory exists + self.data_dir.mkdir(parents=True, exist_ok=True) + + def check_docker(self) -> bool: + """ + Check if Docker is installed and running. + + Returns: + True if Docker is available and running, False otherwise. + """ + docker_path = shutil.which("docker") + if not docker_path: + return False + + try: + # Check if docker command works + result = subprocess.run( + [docker_path, "--version"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode != 0: + return False + + # Check if docker daemon is running + result = subprocess.run( + [docker_path, "info"], + capture_output=True, + text=True, + timeout=10, + ) + return result.returncode == 0 + + except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e: + logger.debug(f"Docker check failed: {e}") + return False + + def require_docker(self) -> str: + """ + Ensure Docker is available, raising an error if not. + + Returns: + Path to docker executable. + + Raises: + DockerNotFoundError: If Docker is not installed or not running. + """ + if self._docker_path: + return self._docker_path + + docker_path = shutil.which("docker") + if not docker_path: + raise DockerNotFoundError( + "Docker is required for sandbox commands.\n" + "Install Docker from https://docs.docker.com/get-docker/" + ) + + # Verify Docker daemon is running + try: + result = subprocess.run( + [docker_path, "info"], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode != 0: + raise DockerNotFoundError( + "Docker daemon is not running.\n" + "Start Docker with: sudo systemctl start docker" + ) + except subprocess.TimeoutExpired: + raise DockerNotFoundError("Docker daemon is not responding.") + except FileNotFoundError: + raise DockerNotFoundError("Docker executable not found.") + + self._docker_path = docker_path + return docker_path + + def _get_container_name(self, sandbox_name: str) -> str: + """Get Docker container name for a sandbox.""" + return f"{self.CONTAINER_PREFIX}{sandbox_name}" + + def _get_metadata_path(self, sandbox_name: str) -> Path: + """Get path to sandbox metadata file.""" + return self.data_dir / f"{sandbox_name}.json" + + def _save_metadata(self, info: SandboxInfo) -> None: + """Save sandbox metadata to disk.""" + metadata_path = self._get_metadata_path(info.name) + with open(metadata_path, "w") as f: + json.dump(info.to_dict(), f, indent=2) + + def _load_metadata(self, sandbox_name: str) -> SandboxInfo | None: + """Load sandbox metadata from disk.""" + metadata_path = self._get_metadata_path(sandbox_name) + if not metadata_path.exists(): + return None + try: + with open(metadata_path) as f: + return SandboxInfo.from_dict(json.load(f)) + except (json.JSONDecodeError, KeyError) as e: + logger.warning(f"Failed to load sandbox metadata: {e}") + return None + + def _delete_metadata(self, sandbox_name: str) -> None: + """Delete sandbox metadata from disk.""" + metadata_path = self._get_metadata_path(sandbox_name) + if metadata_path.exists(): + metadata_path.unlink() + + def _run_docker( + self, + args: list[str], + timeout: int = 60, + check: bool = True, + ) -> subprocess.CompletedProcess: + """ + Run a docker command. + + Args: + args: Arguments to pass to docker (without 'docker' prefix) + timeout: Command timeout in seconds + check: Whether to raise on non-zero exit + + Returns: + CompletedProcess result + """ + docker_path = self.require_docker() + cmd = [docker_path] + args + + logger.debug(f"Running: {' '.join(cmd)}") + + return subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + check=check, + ) + + def create( + self, + name: str, + image: str | None = None, + ) -> SandboxExecutionResult: + """ + Create a new sandbox environment. + + Args: + name: Unique name for the sandbox + image: Docker image to use (default: ubuntu:22.04) + + Returns: + SandboxExecutionResult with success status and message + + Raises: + SandboxAlreadyExistsError: If sandbox with name already exists + DockerNotFoundError: If Docker is not available + """ + self.require_docker() + + # Check if sandbox already exists + existing = self._load_metadata(name) + if existing: + raise SandboxAlreadyExistsError(f"Sandbox '{name}' already exists") + + container_name = self._get_container_name(name) + image = image or self.default_image + + try: + # Pull image if needed + logger.info(f"Pulling image {image}...") + self._run_docker(["pull", image], timeout=300, check=False) + + # Create and start container + logger.info(f"Creating container {container_name}...") + result = self._run_docker( + [ + "run", + "-d", # Detached mode + "--name", + container_name, + "--hostname", + f"sandbox-{name}", + image, + "tail", + "-f", + "/dev/null", # Keep container running + ], + timeout=60, + ) + + container_id = result.stdout.strip()[:12] + + # Update apt cache in container + logger.info("Updating package cache...") + self._run_docker( + ["exec", container_name, "apt-get", "update", "-qq"], + timeout=120, + check=False, + ) + + # Save metadata + info = SandboxInfo( + name=name, + container_id=container_id, + state=SandboxState.RUNNING, + created_at=datetime.now().isoformat(), + image=image, + packages=[], + ) + self._save_metadata(info) + + return SandboxExecutionResult( + success=True, + message=f"Sandbox '{name}' created successfully", + stdout=f"Container ID: {container_id}", + ) + + except subprocess.CalledProcessError as e: + return SandboxExecutionResult( + success=False, + message=f"Failed to create sandbox: {e.stderr}", + exit_code=e.returncode, + stderr=e.stderr, + ) + except subprocess.TimeoutExpired: + return SandboxExecutionResult( + success=False, + message="Timeout while creating sandbox", + exit_code=1, + ) + + def install( + self, + name: str, + package: str, + options: list[str] | None = None, + ) -> SandboxExecutionResult: + """ + Install a package in the sandbox environment. + + Args: + name: Sandbox name + package: Package to install (e.g., "nginx", "docker.io") + options: Additional apt options + + Returns: + SandboxExecutionResult with installation status + """ + self.require_docker() + + # Load sandbox metadata + info = self._load_metadata(name) + if not info: + raise SandboxNotFoundError(f"Sandbox '{name}' not found") + + container_name = self._get_container_name(name) + options = options or [] + + try: + # Install package + apt_cmd = ["apt-get", "install", "-y", "-qq"] + options + [package] + + result = self._run_docker( + ["exec", container_name] + apt_cmd, + timeout=300, + check=False, + ) + + if result.returncode == 0: + # Update metadata with installed package + if package not in info.packages: + info.packages.append(package) + self._save_metadata(info) + + return SandboxExecutionResult( + success=True, + message=f"Package '{package}' installed in sandbox '{name}'", + stdout=result.stdout, + packages_installed=[package], + ) + else: + return SandboxExecutionResult( + success=False, + message=f"Failed to install '{package}': {result.stderr}", + exit_code=result.returncode, + stderr=result.stderr, + ) + + except subprocess.TimeoutExpired: + return SandboxExecutionResult( + success=False, + message=f"Timeout while installing '{package}'", + exit_code=1, + ) + + def test( + self, + name: str, + package: str | None = None, + ) -> SandboxExecutionResult: + """ + Run tests in the sandbox environment. + + Tests include: + - Package functionality (--version, --help) + - Binary existence (which) + - No conflicts detected + + Args: + name: Sandbox name + package: Specific package to test (if None, tests all installed) + + Returns: + SandboxExecutionResult with test results + """ + self.require_docker() + + info = self._load_metadata(name) + if not info: + raise SandboxNotFoundError(f"Sandbox '{name}' not found") + + container_name = self._get_container_name(name) + packages_to_test = [package] if package else info.packages + + if not packages_to_test: + return SandboxExecutionResult( + success=True, + message="No packages to test", + test_results=[], + ) + + test_results: list[SandboxTestResult] = [] + all_passed = True + + for pkg in packages_to_test: + # Test 1: Check if package binary exists + start_time = time.time() + try: + result = self._run_docker( + ["exec", container_name, "which", pkg], + timeout=10, + check=False, + ) + binary_exists = result.returncode == 0 + binary_path = result.stdout.strip() if binary_exists else None + except subprocess.TimeoutExpired: + binary_exists = False + binary_path = None + + if binary_exists: + test_results.append( + SandboxTestResult( + name=f"{pkg}: binary exists", + result=SandboxTestStatus.PASSED, + message=f"Found at {binary_path}", + duration=time.time() - start_time, + ) + ) + else: + # Binary might have different name - check if package is installed + result = self._run_docker( + ["exec", container_name, "dpkg", "-s", pkg], + timeout=10, + check=False, + ) + if result.returncode == 0: + test_results.append( + SandboxTestResult( + name=f"{pkg}: package installed", + result=SandboxTestStatus.PASSED, + message="Package is installed (binary may have different name)", + duration=time.time() - start_time, + ) + ) + else: + test_results.append( + SandboxTestResult( + name=f"{pkg}: package check", + result=SandboxTestStatus.FAILED, + message="Package not found", + duration=time.time() - start_time, + ) + ) + all_passed = False + + # Test 2: Try --version or --help + start_time = time.time() + version_checked = False + + for version_flag in ["--version", "-v", "--help"]: + try: + result = self._run_docker( + ["exec", container_name, pkg, version_flag], + timeout=10, + check=False, + ) + if result.returncode == 0: + test_results.append( + SandboxTestResult( + name=f"{pkg}: functional ({version_flag})", + result=SandboxTestStatus.PASSED, + message=result.stdout[:100].strip(), + duration=time.time() - start_time, + ) + ) + version_checked = True + break + except subprocess.TimeoutExpired: + continue + + if not version_checked and binary_exists: + test_results.append( + SandboxTestResult( + name=f"{pkg}: functional check", + result=SandboxTestStatus.SKIPPED, + message="Could not verify with --version/--help", + duration=time.time() - start_time, + ) + ) + + # Test 3: Check for conflicts (dpkg errors) + start_time = time.time() + result = self._run_docker( + ["exec", container_name, "dpkg", "--audit"], + timeout=30, + check=False, + ) + if result.returncode == 0 and not result.stdout.strip(): + test_results.append( + SandboxTestResult( + name=f"{pkg}: no conflicts", + result=SandboxTestStatus.PASSED, + message="No package conflicts detected", + duration=time.time() - start_time, + ) + ) + elif result.stdout.strip(): + test_results.append( + SandboxTestResult( + name=f"{pkg}: conflict check", + result=SandboxTestStatus.FAILED, + message=result.stdout[:200], + duration=time.time() - start_time, + ) + ) + all_passed = False + + return SandboxExecutionResult( + success=all_passed, + message="All tests passed" if all_passed else "Some tests failed", + test_results=test_results, + ) + + def promote( + self, + name: str, + package: str, + dry_run: bool = False, + ) -> SandboxExecutionResult: + """ + Promote a tested package to the main system. + + This performs a fresh install on the host system (NOT container export). + The sandbox is only used for validation. + + Args: + name: Sandbox name (for validation) + package: Package to install on host + dry_run: If True, show command without executing + + Returns: + SandboxExecutionResult with promotion status + """ + # Verify sandbox exists and package was tested + info = self._load_metadata(name) + if not info: + raise SandboxNotFoundError(f"Sandbox '{name}' not found") + + if package not in info.packages: + return SandboxExecutionResult( + success=False, + message=f"Package '{package}' was not installed in sandbox '{name}'", + exit_code=1, + ) + + # Build the host install command + install_cmd = ["sudo", "apt-get", "install", "-y", package] + + if dry_run: + return SandboxExecutionResult( + success=True, + message=f"Would run: {' '.join(install_cmd)}", + stdout=f"Command: {' '.join(install_cmd)}", + ) + + try: + # Ensure host package lists are fresh before installing + update_cmd = ["sudo", "apt-get", "update", "-qq"] + try: + subprocess.run( + update_cmd, + capture_output=True, + text=True, + timeout=300, + ) + except subprocess.TimeoutExpired: + # If update times out, continue to attempt install but surface warning + logger.warning("Host apt-get update timed out before promote/install") + + # Run apt install on the HOST (not in container) + result = subprocess.run( + install_cmd, + capture_output=True, + text=True, + timeout=300, + ) + + if result.returncode == 0: + return SandboxExecutionResult( + success=True, + message=f"Package '{package}' installed on main system", + stdout=result.stdout, + packages_installed=[package], + ) + else: + # Provide a helpful hint when package cannot be located + hint = "" + combined_output = (result.stderr or "") + "\n" + (result.stdout or "") + if "Unable to locate package" in combined_output: + hint = ( + "\nHint: run 'sudo apt-get update' on the host and retry, " + "or check your APT sources/repositories." + ) + + return SandboxExecutionResult( + success=False, + message=f"Failed to install '{package}' on main system{hint}", + exit_code=result.returncode, + stderr=result.stderr, + ) + + except subprocess.TimeoutExpired: + return SandboxExecutionResult( + success=False, + message="Timeout while installing on main system", + exit_code=1, + ) + + def cleanup(self, name: str, force: bool = False) -> SandboxExecutionResult: + """ + Remove a sandbox environment. + + Args: + name: Sandbox name to remove + force: Force removal even if running + + Returns: + SandboxExecutionResult with cleanup status + """ + self.require_docker() + + container_name = self._get_container_name(name) + + # If metadata is missing, only allow cleanup when forced; otherwise report not found + info = self._load_metadata(name) + if not info and not force: + return SandboxExecutionResult( + success=False, + message=f"Sandbox '{name}' not found", + exit_code=1, + ) + + try: + # Stop container if running (ignore errors) + self._run_docker(["stop", container_name], timeout=30, check=False) + + # Remove container + rm_args = ["rm"] + if force: + rm_args.append("-f") + rm_args.append(container_name) + + self._run_docker(rm_args, timeout=30, check=False) + + # Delete metadata (if exists) + self._delete_metadata(name) + + return SandboxExecutionResult( + success=True, + message=f"Sandbox '{name}' removed", + ) + + except subprocess.TimeoutExpired: + return SandboxExecutionResult( + success=False, + message="Timeout while removing sandbox", + exit_code=1, + ) + + def list_sandboxes(self) -> list[SandboxInfo]: + """ + List all sandbox environments. + + Returns: + List of SandboxInfo objects + """ + sandboxes = [] + + for metadata_file in self.data_dir.glob("*.json"): + try: + with open(metadata_file) as f: + data = json.load(f) + sandboxes.append(SandboxInfo.from_dict(data)) + except (json.JSONDecodeError, KeyError) as e: + logger.warning(f"Failed to load {metadata_file}: {e}") + + return sandboxes + + def get_sandbox(self, name: str) -> SandboxInfo | None: + """ + Get information about a specific sandbox. + + Args: + name: Sandbox name + + Returns: + SandboxInfo or None if not found + """ + return self._load_metadata(name) + + def exec_command( + self, + name: str, + command: list[str], + timeout: int = 60, + ) -> SandboxExecutionResult: + """ + Execute an arbitrary command in the sandbox. + + Args: + name: Sandbox name + command: Command and arguments to execute + timeout: Command timeout in seconds + + Returns: + SandboxExecutionResult with command output + """ + self.require_docker() + + info = self._load_metadata(name) + if not info: + raise SandboxNotFoundError(f"Sandbox '{name}' not found") + + # Check for blocked commands + if command and command[0] in self.SANDBOX_BLOCKED_COMMANDS: + return SandboxExecutionResult( + success=False, + message=f"Command '{command[0]}' is not supported in sandbox", + exit_code=1, + stderr="This command requires system-level access not available in Docker", + ) + + container_name = self._get_container_name(name) + + try: + result = self._run_docker( + ["exec", container_name] + command, + timeout=timeout, + check=False, + ) + + return SandboxExecutionResult( + success=result.returncode == 0, + message="Command executed" if result.returncode == 0 else "Command failed", + exit_code=result.returncode, + stdout=result.stdout, + stderr=result.stderr, + ) + + except subprocess.TimeoutExpired: + return SandboxExecutionResult( + success=False, + message="Command timed out", + exit_code=1, + ) + + @classmethod + def is_sandbox_compatible(cls, command: str) -> tuple[bool, str]: + """ + Check if a command is compatible with sandbox execution. + + Used by LLM command generator to filter incompatible commands. + + Args: + command: Command to check + + Returns: + Tuple of (is_compatible, reason) + """ + # Extract base command + parts = command.split() + if not parts: + return True, "" + + base_cmd = parts[0] + + # Check against blocked commands + if base_cmd in cls.SANDBOX_BLOCKED_COMMANDS: + return False, f"'{base_cmd}' requires system-level access not available in Docker" + + # Check for sudo with blocked commands + if base_cmd == "sudo" and len(parts) > 1: + actual_cmd = parts[1] + if actual_cmd in cls.SANDBOX_BLOCKED_COMMANDS: + return False, f"'{actual_cmd}' requires system-level access not available in Docker" + + return True, "" + + +# Convenience function for checking Docker availability +def docker_available() -> bool: + """Check if Docker is available for sandbox commands.""" + return DockerSandbox().check_docker() diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 173f804a..2637aa98 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -14,6 +14,7 @@ This document provides a comprehensive reference for all commands available in t | `cortex history` | View installation history | | `cortex rollback ` | Undo an installation | | `cortex stack ` | Install a pre-built package stack | +| `cortex sandbox ` | Test packages in Docker sandbox | | `cortex cache stats` | Show LLM cache statistics | | `cortex notify` | Manage desktop notifications | @@ -261,6 +262,63 @@ cortex cache stats --- +### `cortex sandbox` + +Test packages in isolated Docker containers before installing to the main system. Requires Docker. + +**Usage:** +```bash +cortex sandbox [options] +``` + +**Actions:** +| Action | Description | +|--------|-------------| +| `create ` | Create a sandbox environment | +| `install ` | Install package in sandbox | +| `test [pkg]` | Run automated tests in sandbox | +| `promote ` | Install tested package on main system | +| `cleanup ` | Remove sandbox environment | +| `list` | List all sandbox environments | +| `exec ` | Execute command in sandbox | + +**Options:** +| Flag | Description | +|------|-------------| +| `--image ` | Docker image for create (default: ubuntu:22.04) | +| `--dry-run` | Preview promote without executing | +| `-y, --yes` | Skip confirmation for promote | +| `-f, --force` | Force cleanup even if running | + +**Examples:** +```bash +# Create a sandbox +cortex sandbox create test-env + +# Install package in sandbox +cortex sandbox install test-env nginx + +# Run tests +cortex sandbox test test-env + +# Promote to main system +cortex sandbox promote test-env nginx + +# Cleanup +cortex sandbox cleanup test-env + +# Use custom base image +cortex sandbox create debian-test --image debian:12 +``` + +**Notes:** +- Docker must be installed and running +- Promotion runs fresh `apt install` on host (not container export) +- Some commands (`systemctl`, `service`) are blocked in sandbox +- See [SANDBOX.md](SANDBOX.md) for full documentation + +--- + ### `cortex notify` Manage desktop notification settings for installation events. diff --git a/docs/SANDBOX.md b/docs/SANDBOX.md new file mode 100644 index 00000000..fd927e64 --- /dev/null +++ b/docs/SANDBOX.md @@ -0,0 +1,296 @@ +# Package Sandbox Testing Environment + +Test packages in isolated Docker containers before installing to the main system. + +## Overview + +The sandbox feature provides a safe way to test packages before committing to a system-wide installation. It uses Docker containers as isolated environments, allowing you to: + +- Install packages without affecting your system +- Run automated tests to validate functionality +- Safely promote tested packages to your main system +- Clean up test environments when done + +## Requirements + +- **Docker** is required for sandbox commands +- Docker must be installed and running +- Other Cortex commands work without Docker + +```bash +# Check if Docker is available +docker --version +docker info +``` + +If Docker is not installed, sandbox commands will show: +``` +Error: Docker is required for sandbox commands. +Install Docker from https://docs.docker.com/get-docker/ +``` + +## Quick Start + +```bash +# Create a sandbox environment +$ cortex sandbox create test-env +āœ“ Sandbox environment 'test-env' created + +# Install a package in the sandbox +$ cortex sandbox install test-env nginx +āœ“ nginx installed in sandbox + +# Run tests to validate the package +$ cortex sandbox test test-env +Running tests in sandbox 'test-env'... + āœ“ nginx: binary exists + āœ“ nginx: functional (--version) + āœ“ nginx: no conflicts +All tests passed + +# Promote to main system (fresh install on host) +$ cortex sandbox promote test-env nginx +Promote 'nginx' to main system? [Y/n]: y +āœ“ nginx installed on main system + +# Clean up the sandbox +$ cortex sandbox cleanup test-env +āœ“ Sandbox 'test-env' removed +``` + +## Commands + +### `cortex sandbox create ` + +Create a new sandbox environment. + +```bash +cortex sandbox create test-env +cortex sandbox create test-env --image debian:12 +``` + +Options: +- `--image`: Docker image to use (default: `ubuntu:22.04`) + +### `cortex sandbox install ` + +Install a package in the sandbox environment. + +```bash +cortex sandbox install test-env nginx +cortex sandbox install test-env docker.io +``` + +### `cortex sandbox test [package]` + +Run automated tests in the sandbox. + +```bash +# Test all installed packages +cortex sandbox test test-env + +# Test a specific package +cortex sandbox test test-env nginx +``` + +Tests performed: +- **Binary exists**: Checks if package binary is available +- **Functional**: Runs `--version` or `--help` to verify it works +- **No conflicts**: Runs `dpkg --audit` to check for issues + +### `cortex sandbox promote ` + +Install a tested package on the main system. + +```bash +cortex sandbox promote test-env nginx +cortex sandbox promote test-env nginx --dry-run # Preview only +cortex sandbox promote test-env nginx -y # Skip confirmation +``` + +Options: +- `--dry-run`: Show what would be installed without executing +- `-y, --yes`: Skip the confirmation prompt + +**Important**: Promotion runs a fresh `apt install` on the host system. It does NOT copy files from the container. + +### `cortex sandbox cleanup ` + +Remove a sandbox environment. + +```bash +cortex sandbox cleanup test-env +cortex sandbox cleanup test-env --force +``` + +Options: +- `-f, --force`: Force removal even if container is running + +### `cortex sandbox list` + +List all sandbox environments. + +```bash +cortex sandbox list +``` + +### `cortex sandbox exec ` + +Execute an arbitrary command in the sandbox. + +```bash +cortex sandbox exec test-env cat /etc/os-release +cortex sandbox exec test-env apt list --installed +``` + +## Workflow Example + +### Testing a Complex Installation + +```bash +# Create isolated environment +cortex sandbox create docker-test + +# Install Docker in sandbox +cortex sandbox install docker-test docker.io + +# Run tests +cortex sandbox test docker-test +# Output: +# āœ“ docker.io: package installed +# āœ“ docker.io: no conflicts + +# Check version +cortex sandbox exec docker-test docker --version + +# If satisfied, promote to main system +cortex sandbox promote docker-test docker.io + +# Clean up +cortex sandbox cleanup docker-test +``` + +### Testing Multiple Packages + +```bash +cortex sandbox create webstack + +cortex sandbox install webstack nginx +cortex sandbox install webstack postgresql +cortex sandbox install webstack redis-server + +cortex sandbox test webstack + +# Promote all +cortex sandbox promote webstack nginx -y +cortex sandbox promote webstack postgresql -y +cortex sandbox promote webstack redis-server -y + +cortex sandbox cleanup webstack +``` + +## Limitations + +Sandbox environments run inside Docker containers. The following **cannot** be tested in sandbox: + +| Category | Examples | Reason | +|----------|----------|--------| +| System services | `systemctl start nginx` | No systemd in container | +| Kernel modules | `modprobe`, `insmod` | Shared host kernel | +| Hardware access | `nvidia-smi`, `lspci` | No device passthrough | +| Network config | `iptables`, `ufw` | Network namespace isolation | +| Boot/init | `reboot`, `shutdown` | Container lifecycle | + +### What Works Well + +āœ… Package installation (`apt install`) +āœ… Binary version checks (`--version`) +āœ… Basic functionality tests +āœ… Dependency resolution +āœ… Conflict detection + +### What Doesn't Work + +āŒ Service management (`systemctl`, `service`) +āŒ Kernel operations (`modprobe`, `sysctl`) +āŒ Hardware detection and drivers +āŒ Network firewall rules +āŒ System boot behavior + +## How Promotion Works + +When you run `cortex sandbox promote`, Cortex does **NOT**: +- Export Docker container filesystem +- Copy files from container to host +- Use any Docker layer magic + +Instead, it simply runs: +```bash +sudo apt-get install -y +``` + +The sandbox is a **validation step only**. Think of it as a "dry run with real installation" to catch issues before they affect your system. + +## Data Storage + +Sandbox metadata is stored in `~/.cortex/sandboxes/`: +``` +~/.cortex/sandboxes/ +ā”œā”€ā”€ test-env.json +ā”œā”€ā”€ webstack.json +ā””ā”€ā”€ docker-test.json +``` + +Each file contains: +- Sandbox name +- Container ID +- Docker image used +- Creation timestamp +- List of installed packages + +## Troubleshooting + +### Docker permission denied + +```bash +# Add your user to the docker group +sudo usermod -aG docker $USER +# Log out and back in +``` + +### Container won't start + +```bash +# Check if Docker daemon is running +sudo systemctl status docker + +# Start Docker if needed +sudo systemctl start docker +``` + +### Sandbox metadata out of sync + +If a container was removed manually: +```bash +# Force cleanup metadata +rm ~/.cortex/sandboxes/.json +``` + +### Tests failing unexpectedly + +Some packages don't have binaries matching their package name: +```bash +# Use exec to investigate +cortex sandbox exec test-env dpkg -L +cortex sandbox exec test-env apt show +``` + +## Integration with Cortex + +The sandbox feature integrates with Cortex's AI command generation. When running in sandbox mode, the LLM is constrained to: + +- Use only `apt` for installation +- Avoid `systemctl`, `service`, and kernel commands +- Prefer version checks (`--version`) for validation + +This ensures generated commands are compatible with the Docker sandbox environment. diff --git a/tests/test_docker_sandbox.py b/tests/test_docker_sandbox.py new file mode 100644 index 00000000..e98bf346 --- /dev/null +++ b/tests/test_docker_sandbox.py @@ -0,0 +1,438 @@ +#!/usr/bin/env python3 +""" +Tests for Docker-based Package Sandbox Testing Environment. + +Tests cover: +- DockerSandbox class methods (create, install, test, promote, cleanup) +- Docker detection and error handling +- Edge cases and error conditions +""" + +import json +import os +import shutil as shutil_module +import sys +import tempfile +import unittest +from pathlib import Path +from typing import Any +from unittest.mock import Mock, patch + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from cortex.sandbox.docker_sandbox import ( + DockerNotFoundError, + DockerSandbox, + SandboxAlreadyExistsError, + SandboxInfo, + SandboxNotFoundError, + SandboxState, + SandboxTestStatus, + docker_available, +) + + +def create_sandbox_metadata( + name: str = "test-env", + packages: list[str] | None = None, + state: str = "running", +) -> dict[str, Any]: + """Create a sandbox metadata dictionary.""" + return { + "name": name, + "container_id": f"abc123{name}", + "state": state, + "created_at": "2024-01-01T00:00:00", + "image": "ubuntu:22.04", + "packages": packages or [], + } + + +def mock_docker_available() -> tuple[str, Mock]: + """Return mocks configured for Docker available.""" + return "/usr/bin/docker", Mock(returncode=0, stdout="Docker info", stderr="") + + +class SandboxTestBase(unittest.TestCase): + """Base class for sandbox tests with common setup/teardown.""" + + def setUp(self) -> None: + """Set up temp directory for sandbox metadata.""" + self.temp_dir = tempfile.mkdtemp() + self.data_dir = Path(self.temp_dir) / "sandboxes" + self.data_dir.mkdir(parents=True, exist_ok=True) + + def tearDown(self) -> None: + """Clean up temp directory.""" + shutil_module.rmtree(self.temp_dir, ignore_errors=True) + + def write_metadata( + self, + name: str = "test-env", + packages: list[str] | None = None, + state: str = "running", + ) -> dict[str, Any]: + """Helper to write sandbox metadata to disk.""" + metadata = create_sandbox_metadata(name, packages, state) + with open(self.data_dir / f"{name}.json", "w") as f: + json.dump(metadata, f) + return metadata + + def create_sandbox_instance(self) -> DockerSandbox: + """Create a DockerSandbox instance with test data directory.""" + return DockerSandbox(data_dir=self.data_dir) + + +class TestDockerDetection(unittest.TestCase): + """Tests for Docker availability detection.""" + + @patch("shutil.which") + def test_docker_not_installed(self, mock_which: Mock) -> None: + """Test detection when Docker is not installed.""" + mock_which.return_value = None + self.assertFalse(DockerSandbox().check_docker()) + + @patch("shutil.which") + @patch("subprocess.run") + def test_docker_installed_but_not_running(self, mock_run: Mock, mock_which: Mock) -> None: + """Test detection when Docker is installed but daemon not running.""" + mock_which.return_value = "/usr/bin/docker" + mock_run.side_effect = [ + Mock(returncode=0, stdout="Docker version 24.0.0"), + Mock(returncode=1, stderr="Cannot connect to Docker daemon"), + ] + self.assertFalse(DockerSandbox().check_docker()) + + @patch("shutil.which") + @patch("subprocess.run") + def test_docker_available(self, mock_run: Mock, mock_which: Mock) -> None: + """Test detection when Docker is fully available.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + self.assertTrue(DockerSandbox().check_docker()) + + @patch("shutil.which") + def test_require_docker_raises_when_not_found(self, mock_which: Mock) -> None: + """Test require_docker raises DockerNotFoundError when not installed.""" + mock_which.return_value = None + with self.assertRaises(DockerNotFoundError) as ctx: + DockerSandbox().require_docker() + self.assertIn("Docker is required", str(ctx.exception)) + + @patch("shutil.which") + @patch("subprocess.run") + def test_require_docker_raises_when_daemon_not_running( + self, mock_run: Mock, mock_which: Mock + ) -> None: + """Test require_docker raises when daemon not running.""" + mock_which.return_value = "/usr/bin/docker" + mock_run.return_value = Mock(returncode=1, stderr="Cannot connect") + with self.assertRaises(DockerNotFoundError) as ctx: + DockerSandbox().require_docker() + self.assertIn("not running", str(ctx.exception)) + + +class TestSandboxCreate(SandboxTestBase): + """Tests for sandbox creation.""" + + @patch("shutil.which") + @patch("subprocess.run") + def test_create_sandbox_success(self, mock_run: Mock, mock_which: Mock) -> None: + """Test successful sandbox creation.""" + mock_which.return_value, _ = mock_docker_available() + mock_run.return_value = Mock(returncode=0, stdout="abc123def456", stderr="") + + result = self.create_sandbox_instance().create("test-env") + + self.assertTrue(result.success) + self.assertIn("test-env", result.message) + self.assertTrue((self.data_dir / "test-env.json").exists()) + + @patch("shutil.which") + @patch("subprocess.run") + def test_create_sandbox_already_exists(self, mock_run: Mock, mock_which: Mock) -> None: + """Test error when sandbox already exists.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + sandbox = self.create_sandbox_instance() + sandbox.create("test-env") + + with self.assertRaises(SandboxAlreadyExistsError): + sandbox.create("test-env") + + @patch("shutil.which") + @patch("subprocess.run") + def test_create_sandbox_with_custom_image(self, mock_run: Mock, mock_which: Mock) -> None: + """Test sandbox creation with custom image.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + sandbox = self.create_sandbox_instance() + sandbox.create("test-env", image="debian:12") + + self.assertEqual(sandbox.get_sandbox("test-env").image, "debian:12") + + +class TestSandboxInstall(SandboxTestBase): + """Tests for package installation in sandbox.""" + + def setUp(self) -> None: + super().setUp() + self.write_metadata("test-env", packages=[]) + + @patch("shutil.which") + @patch("subprocess.run") + def test_install_package_success(self, mock_run: Mock, mock_which: Mock) -> None: + """Test successful package installation.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + result = self.create_sandbox_instance().install("test-env", "nginx") + + self.assertTrue(result.success) + self.assertIn("nginx", result.packages_installed) + + @patch("shutil.which") + @patch("subprocess.run") + def test_install_package_failure(self, mock_run: Mock, mock_which: Mock) -> None: + """Test package installation failure.""" + mock_which.return_value = "/usr/bin/docker" + mock_run.side_effect = [ + Mock(returncode=0, stdout="Docker info", stderr=""), + Mock(returncode=100, stdout="", stderr="E: Unable to locate package"), + ] + + result = self.create_sandbox_instance().install("test-env", "nonexistent") + + self.assertFalse(result.success) + self.assertIn("Failed to install", result.message) + + @patch("shutil.which") + @patch("subprocess.run") + def test_install_sandbox_not_found(self, mock_run: Mock, mock_which: Mock) -> None: + """Test installation in non-existent sandbox.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + with self.assertRaises(SandboxNotFoundError): + self.create_sandbox_instance().install("nonexistent", "nginx") + + +class TestSandboxTest(SandboxTestBase): + """Tests for sandbox testing functionality.""" + + def setUp(self) -> None: + super().setUp() + self.write_metadata("test-env", packages=["nginx"]) + + @patch("shutil.which") + @patch("subprocess.run") + def test_test_all_pass(self, mock_run: Mock, mock_which: Mock) -> None: + """Test when all tests pass.""" + mock_which.return_value = "/usr/bin/docker" + mock_run.side_effect = [ + Mock(returncode=0, stdout="/usr/bin/docker"), + Mock(returncode=0, stdout="/usr/sbin/nginx"), + Mock(returncode=0, stdout="nginx version: 1.18"), + Mock(returncode=0, stdout=""), + ] + + result = self.create_sandbox_instance().test("test-env") + + self.assertTrue(result.success) + passed = [t for t in result.test_results if t.result == SandboxTestStatus.PASSED] + self.assertTrue(len(passed) > 0) + + @patch("shutil.which") + @patch("subprocess.run") + def test_test_no_packages(self, mock_run: Mock, mock_which: Mock) -> None: + """Test when no packages installed.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + self.write_metadata("empty-env", packages=[]) + + result = self.create_sandbox_instance().test("empty-env") + + self.assertTrue(result.success) + self.assertEqual(len(result.test_results), 0) + + +# ============================================================================= +# Sandbox Promote Tests +# ============================================================================= + + +class TestSandboxPromote(SandboxTestBase): + """Tests for package promotion to main system.""" + + def setUp(self) -> None: + super().setUp() + self.write_metadata("test-env", packages=["nginx"]) + + @patch("subprocess.run") + def test_promote_dry_run(self, mock_run: Mock) -> None: + """Test promotion in dry-run mode.""" + result = self.create_sandbox_instance().promote("test-env", "nginx", dry_run=True) + + self.assertTrue(result.success) + self.assertIn("Would run", result.message) + + def test_promote_package_not_in_sandbox(self) -> None: + """Test promotion of package not installed in sandbox.""" + result = self.create_sandbox_instance().promote("test-env", "redis", dry_run=False) + + self.assertFalse(result.success) + self.assertIn("not installed in sandbox", result.message) + + @patch("subprocess.run") + def test_promote_success(self, mock_run: Mock) -> None: + """Test successful promotion.""" + mock_run.return_value = Mock(returncode=0, stdout="", stderr="") + + result = self.create_sandbox_instance().promote("test-env", "nginx", dry_run=False) + + self.assertTrue(result.success) + call_args = mock_run.call_args[0][0] + self.assertEqual(call_args, ["sudo", "apt-get", "install", "-y", "nginx"]) + + +class TestSandboxCleanup(SandboxTestBase): + """Tests for sandbox cleanup.""" + + def setUp(self) -> None: + super().setUp() + self.write_metadata("test-env") + + @patch("shutil.which") + @patch("subprocess.run") + def test_cleanup_success(self, mock_run: Mock, mock_which: Mock) -> None: + """Test successful cleanup.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + result = self.create_sandbox_instance().cleanup("test-env") + + self.assertTrue(result.success) + self.assertFalse((self.data_dir / "test-env.json").exists()) + + @patch("shutil.which") + @patch("subprocess.run") + def test_cleanup_force(self, mock_run: Mock, mock_which: Mock) -> None: + """Test force cleanup.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + result = self.create_sandbox_instance().cleanup("test-env", force=True) + + self.assertTrue(result.success) + + +class TestSandboxList(SandboxTestBase): + """Tests for listing sandboxes.""" + + def test_list_empty(self) -> None: + """Test listing when no sandboxes exist.""" + self.assertEqual(len(self.create_sandbox_instance().list_sandboxes()), 0) + + def test_list_multiple(self) -> None: + """Test listing multiple sandboxes.""" + for name in ["env1", "env2", "env3"]: + self.write_metadata(name) + + sandboxes = self.create_sandbox_instance().list_sandboxes() + + self.assertEqual(len(sandboxes), 3) + self.assertEqual({s.name for s in sandboxes}, {"env1", "env2", "env3"}) + + +class TestSandboxExec(SandboxTestBase): + """Tests for command execution in sandbox.""" + + def setUp(self) -> None: + super().setUp() + self.write_metadata("test-env") + + @patch("shutil.which") + @patch("subprocess.run") + def test_exec_success(self, mock_run: Mock, mock_which: Mock) -> None: + """Test successful command execution.""" + mock_which.return_value = "/usr/bin/docker" + mock_run.return_value = Mock(returncode=0, stdout="Hello\n", stderr="") + + result = self.create_sandbox_instance().exec_command("test-env", ["echo", "Hello"]) + + self.assertTrue(result.success) + self.assertIn("Hello", result.stdout) + + @patch("shutil.which") + @patch("subprocess.run") + def test_exec_blocked_command(self, mock_run: Mock, mock_which: Mock) -> None: + """Test blocked command is rejected.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + + result = self.create_sandbox_instance().exec_command( + "test-env", ["systemctl", "start", "nginx"] + ) + + self.assertFalse(result.success) + self.assertIn("not supported", result.message) + + +class TestSandboxCompatibility(unittest.TestCase): + """Tests for command compatibility checking.""" + + def test_allowed_commands(self) -> None: + """Test that normal commands are allowed.""" + self.assertTrue(DockerSandbox.is_sandbox_compatible("apt install nginx")[0]) + self.assertTrue(DockerSandbox.is_sandbox_compatible("nginx --version")[0]) + + def test_blocked_commands(self) -> None: + """Test that blocked commands are rejected.""" + is_compat, reason = DockerSandbox.is_sandbox_compatible("systemctl start nginx") + self.assertFalse(is_compat) + self.assertIn("systemctl", reason) + + self.assertFalse(DockerSandbox.is_sandbox_compatible("sudo service nginx restart")[0]) + self.assertFalse(DockerSandbox.is_sandbox_compatible("modprobe loop")[0]) + + +class TestSandboxInfo(unittest.TestCase): + """Tests for SandboxInfo data class.""" + + def test_to_dict(self) -> None: + """Test conversion to dictionary.""" + info = SandboxInfo( + name="test", + container_id="abc123", + state=SandboxState.RUNNING, + created_at="2024-01-01T00:00:00", + image="ubuntu:22.04", + packages=["nginx", "redis"], + ) + data = info.to_dict() + + self.assertEqual(data["name"], "test") + self.assertEqual(data["state"], "running") + self.assertEqual(data["packages"], ["nginx", "redis"]) + + def test_from_dict(self) -> None: + """Test creation from dictionary.""" + info = SandboxInfo.from_dict(create_sandbox_metadata("test", ["nginx"])) + + self.assertEqual(info.name, "test") + self.assertEqual(info.state, SandboxState.RUNNING) + self.assertIn("nginx", info.packages) + + +class TestDockerAvailableFunction(unittest.TestCase): + """Tests for docker_available() convenience function.""" + + @patch("shutil.which") + @patch("subprocess.run") + def test_docker_available_true(self, mock_run: Mock, mock_which: Mock) -> None: + """Test when Docker is available.""" + mock_which.return_value, mock_run.return_value = mock_docker_available() + self.assertTrue(docker_available()) + + @patch("shutil.which") + def test_docker_available_false(self, mock_which: Mock) -> None: + """Test when Docker is not available.""" + mock_which.return_value = None + self.assertFalse(docker_available()) + + +if __name__ == "__main__": + unittest.main()