""" Binary manager for downloading and managing required tools. """ import os import sys import subprocess import shutil import platform import urllib.request import tarfile import tempfile from pathlib import Path from typing import Optional, List, Dict from .utils import Logger, Colors class BinaryManager: """Manages required binaries (download and execute).""" # Binary versions VERSIONS = { 'helm': 'v3.17.1', 'minikube': 'v1.37.0', 'kubectl': 'v1.28.4', 'jsonnet': 'v0.21.0', } # Download URLs by platform DOWNLOAD_URLS = { 'helm': { 'linux-amd64': 'https://get.helm.sh/helm-{version}-linux-amd64.tar.gz', 'linux-arm64': 'https://get.helm.sh/helm-{version}-linux-arm64.tar.gz', 'darwin-amd64': 'https://get.helm.sh/helm-{version}-darwin-amd64.tar.gz', 'darwin-arm64': 'https://get.helm.sh/helm-{version}-darwin-arm64.tar.gz', }, 'minikube': { 'linux-amd64': 'https://github.com/kubernetes/minikube/releases/download/{version}/minikube-linux-amd64', 'linux-arm64': 'https://github.com/kubernetes/minikube/releases/download/{version}/minikube-linux-arm64', 'darwin-amd64': 'https://github.com/kubernetes/minikube/releases/download/{version}/minikube-darwin-amd64', 'darwin-arm64': 'https://github.com/kubernetes/minikube/releases/download/{version}/minikube-darwin-arm64', }, 'kubectl': { 'linux-amd64': 'https://dl.k8s.io/release/{version}/bin/linux/amd64/kubectl', 'linux-arm64': 'https://dl.k8s.io/release/{version}/bin/linux/arm64/kubectl', 'darwin-amd64': 'https://dl.k8s.io/release/{version}/bin/darwin/amd64/kubectl', 'darwin-arm64': 'https://dl.k8s.io/release/{version}/bin/darwin/arm64/kubectl', }, 'jsonnet': { 'linux-amd64': 'https://github.com/google/go-jsonnet/releases/download/{version}/go-jsonnet_Linux_x86_64.tar.gz', 'linux-arm64': 'https://github.com/google/go-jsonnet/releases/download/{version}/go-jsonnet_Linux_arm64.tar.gz', 'darwin-amd64': 'https://github.com/google/go-jsonnet/releases/download/{version}/go-jsonnet_Darwin_x86_64.tar.gz', 'darwin-arm64': 'https://github.com/google/go-jsonnet/releases/download/{version}/go-jsonnet_Darwin_arm64.tar.gz', } } def __init__(self, tiop_home: Path): self.tiop_home = tiop_home self.bin_dir = tiop_home / 'bin' self.bin_dir.mkdir(parents=True, exist_ok=True) # Detect platform system = platform.system().lower() if system == 'linux': self.os = 'linux' elif system == 'darwin': self.os = 'darwin' else: Logger.die(f"Unsupported operating system: {system}") # Detect architecture machine = platform.machine().lower() if machine in ['x86_64', 'amd64']: self.arch = 'amd64' elif machine in ['aarch64', 'arm64']: self.arch = 'arm64' else: Logger.die(f"Unsupported architecture: {machine}") self.platform = f"{self.os}-{self.arch}" def get_binary_path(self, name: str) -> Path: """Get the path to a binary.""" return self.bin_dir / name def is_binary_available(self, name: str) -> bool: """Check if a binary is available (either in bin_dir or in PATH).""" # Check in bin_dir first (preferred) binary_path = self.get_binary_path(name) if binary_path.exists() and os.access(binary_path, os.X_OK): return True # Check in PATH as fallback if shutil.which(name): return True return False def get_binary_command(self, name: str) -> str: """Get the command to execute a binary.""" # Prefer local binary first local_binary = self.get_binary_path(name) if local_binary.exists() and os.access(local_binary, os.X_OK): return str(local_binary) # Fall back to system binary system_binary = shutil.which(name) if system_binary: return system_binary # Not found anywhere return str(local_binary) def _download_progress_hook(self, block_num, block_size, total_size): """Progress hook for urllib.request.urlretrieve.""" downloaded = block_num * block_size if total_size > 0: percent = min(downloaded * 100 / total_size, 100) downloaded_mb = downloaded / (1024 * 1024) total_mb = total_size / (1024 * 1024) # Create progress bar bar_length = 40 filled_length = int(bar_length * downloaded / total_size) bar = '=' * filled_length + '-' * (bar_length - filled_length) # Print progress with carriage return to overwrite line sys.stdout.write(f'\r[{bar}] {percent:.1f}% ({downloaded_mb:.1f}/{total_mb:.1f} MB)') sys.stdout.flush() if downloaded >= total_size: sys.stdout.write('\n') sys.stdout.flush() else: # Unknown size, just show downloaded amount downloaded_mb = downloaded / (1024 * 1024) sys.stdout.write(f'\rDownloaded: {downloaded_mb:.1f} MB') sys.stdout.flush() def download_binary(self, name: str, force: bool = False): """Download a binary if not already available.""" if not force and self.is_binary_available(name): Logger.info(f"{name} is already available") return if name not in self.DOWNLOAD_URLS: Logger.die(f"Unknown binary: {name}") if self.platform not in self.DOWNLOAD_URLS[name]: Logger.die(f"No download URL available for {name} on {self.platform}") version = self.VERSIONS[name] url = self.DOWNLOAD_URLS[name][self.platform].format(version=version) Logger.info(f"Downloading {name} {version} for {self.platform}...") Logger.info(f"URL: {url}") try: with tempfile.TemporaryDirectory() as tmpdir: tmpdir_path = Path(tmpdir) # Download file if url.endswith('.tar.gz'): archive_path = tmpdir_path / f"{name}.tar.gz" urllib.request.urlretrieve(url, archive_path, reporthook=self._download_progress_hook) # Extract Logger.info(f"Extracting {name}...") with tarfile.open(archive_path, 'r:gz') as tar: tar.extractall(tmpdir_path) # Find the binary if name == 'helm': # Helm extracts to linux-amd64/helm or similar extracted_binary = list(tmpdir_path.glob(f"*/helm"))[0] elif name == 'jsonnet': # Jsonnet extracts jsonnet directly extracted_binary = tmpdir_path / 'jsonnet' else: extracted_binary = tmpdir_path / name # Copy to bin_dir shutil.copy2(extracted_binary, self.get_binary_path(name)) else: # Direct binary download urllib.request.urlretrieve(url, self.get_binary_path(name), reporthook=self._download_progress_hook) # Make executable os.chmod(self.get_binary_path(name), 0o755) Logger.success(f"{name} {version} installed successfully") except Exception as e: Logger.die(f"Failed to download {name}: {e}") def ensure_binary(self, name: str): """Ensure a binary is available, download if necessary.""" if not self.is_binary_available(name): Logger.info(f"{name} not found, downloading...") self.download_binary(name) def ensure_all_binaries(self): """Ensure all required binaries are available.""" for name in ['helm', 'minikube', 'kubectl', 'jsonnet']: self.ensure_binary(name) def run(self, name: str, args: List[str], env: Optional[Dict[str, str]] = None, capture_output: bool = False, check: bool = True, input=None) -> subprocess.CompletedProcess: """Run a binary with the given arguments.""" self.ensure_binary(name) cmd = [self.get_binary_command(name)] + args run_env = os.environ.copy() if env: run_env.update(env) Logger.trace(f"> {Colors.GREEN}{name}{Colors.NC} args={args} env={env}") if capture_output: return subprocess.run(cmd, env=run_env, capture_output=True, text=True, check=check, input=input) else: return subprocess.run(cmd, env=run_env, check=check, input=input)