openenv-testing/echo_env

1#!/usr/bin/env python32"""3Git Server Client for connecting to external Gitea instance.4 5This module provides a lightweight client for interacting with a shared6Gitea service, optimized for task-based isolation where multiple environment7instances share the same Gitea server but have isolated workspaces.8"""9 10import json11import os12import shutil13import subprocess14import time15from dataclasses import dataclass16from pathlib import Path17from urllib.parse import urlparse18 19 20@dataclass21class RepoInfo:22    """Information about a repository."""23 24    name: str25    url: str26    commit: str27    clone_url: str28 29 30class GitServerClient:31    """32    Client for connecting to an external Gitea server.33 34    This client is optimized for task-based isolation where:35    - Multiple tasks share the same Gitea instance36    - Each task has its own isolated workspace37    - Fast reset() via git operations (no server restart)38    - Repos are pre-migrated to Gitea once39 40    Args:41        gitea_url: URL of the Gitea server (e.g., "http://gitea:3000")42        username: Gitea username for authentication43        password: Gitea password for authentication44        workspace_dir: Local workspace directory for cloning repos45 46    Example:47        >>> # Connect to shared Gitea (credentials from environment)48        >>> import os49        >>> client = GitServerClient(50        ...     gitea_url=os.getenv("GITEA_URL"),51        ...     username=os.getenv("GITEA_USERNAME"),52        ...     password=os.getenv("GITEA_PASSWORD")53        ... )54        >>> client.wait_for_ready()55        >>> # Clone repo to workspace56        >>> path = client.clone_to_workspace("my-repo", commit="abc123")57        >>> # Fast reset to base state58        >>> client.reset_workspace("my-repo", commit="abc123")59    """60 61    def __init__(62        self,63        gitea_url: str,64        username: str,65        password: str,66        workspace_dir: str = "/workspace",67    ):68        """Initialize Git Server Client."""69        self.gitea_url = gitea_url.rstrip("/")70        self.username = username71        self.password = password72        self.workspace_dir = Path(workspace_dir)73        self.is_ready = False74 75        # Parse Gitea URL76        parsed = urlparse(self.gitea_url)77        self.domain = parsed.hostname or "localhost"78        self.port = parsed.port or 300079 80        # Ensure workspace exists81        os.makedirs(self.workspace_dir, exist_ok=True)82 83        # Configure git credentials84        self._configure_git()85 86    def _configure_git(self):87        """Configure git credentials for automatic authentication."""88        home_dir = Path.home()89 90        # Git config91        git_config = f"""[user]92    name = {self.username}93    email = {self.username}@local.env94[init]95    defaultBranch = main96[credential]97    helper = store98"""99        gitconfig_path = home_dir / ".gitconfig"100        gitconfig_path.write_text(git_config)101 102        # Git credentials103        git_credentials = f"http://{self.username}:{self.password}@{self.domain}:{self.port}\n"104        gitcreds_path = home_dir / ".git-credentials"105        gitcreds_path.write_text(git_credentials)106        gitcreds_path.chmod(0o600)107 108    def wait_for_ready(self, timeout: int = 30) -> bool:109        """110        Wait for Gitea server to be ready.111 112        Args:113            timeout: Maximum seconds to wait114 115        Returns:116            True if server is ready, False otherwise117        """118        start_time = time.time()119        while time.time() - start_time < timeout:120            try:121                result = subprocess.run(122                    ["curl", "-sf", f"{self.gitea_url}/"],123                    capture_output=True,124                    timeout=5,125                )126                if result.returncode == 0:127                    self.is_ready = True128                    return True129            except subprocess.TimeoutExpired:130                pass131            except Exception:132                pass133 134            time.sleep(1)135 136        return False137 138    def list_repositories(self) -> list[dict[str, str]]:139        """140        List all repositories in Gitea.141 142        Returns:143            List of repository information dictionaries144        """145        if not self.is_ready:146            raise RuntimeError("Gitea server is not ready")147 148        result = subprocess.run(149            [150                "curl",151                "-s",152                f"{self.gitea_url}/api/v1/user/repos",153                "-u",154                f"{self.username}:{self.password}",155            ],156            capture_output=True,157            text=True,158        )159 160        if result.returncode != 0:161            return []162 163        try:164            repos = json.loads(result.stdout)165            return [166                {167                    "name": repo["name"],168                    "full_name": repo["full_name"],169                    "clone_url": repo["clone_url"],170                    "description": repo.get("description", ""),171                }172                for repo in repos173            ]174        except (json.JSONDecodeError, KeyError):175            return []176 177    def clone_to_workspace(178        self, repo_name: str, target_dir: str | None = None, commit: str = "main"179    ) -> str:180        """181        Clone a repository to the workspace at a specific commit.182 183        This creates a fresh clone optimized for task isolation.184 185        Args:186            repo_name: Name of repository to clone187            target_dir: Target directory name (defaults to repo_name)188            commit: Commit hash or branch to checkout189 190        Returns:191            Path to cloned repository192 193        Raises:194            RuntimeError: If clone fails195        """196        if not self.is_ready:197            raise RuntimeError("Gitea server is not ready")198 199        target_dir = target_dir or repo_name200        target_path = self.workspace_dir / target_dir201 202        # Remove existing directory if present203        if target_path.exists():204            shutil.rmtree(target_path)205 206        clone_url = f"{self.gitea_url}/{self.username}/{repo_name}.git"207 208        # Clone repository209        result = subprocess.run(210            ["git", "clone", clone_url, str(target_path)],211            capture_output=True,212            text=True,213        )214 215        if result.returncode != 0:216            raise RuntimeError(f"Clone failed: {result.stderr}")217 218        # Checkout specific commit219        if commit != "main":220            result = subprocess.run(221                ["git", "checkout", commit],222                cwd=str(target_path),223                capture_output=True,224                text=True,225            )226 227            if result.returncode != 0:228                raise RuntimeError(f"Checkout failed: {result.stderr}")229 230        return str(target_path)231 232    def reset_workspace(self, repo_name: str, commit: str = "main") -> bool:233        """234        Fast reset of workspace to base state (optimized for task resets).235 236        This is much faster than re-cloning. It:237        1. Checks out the target commit238        2. Resets to that commit (hard)239        3. Cleans untracked files240 241        Args:242            repo_name: Name of repository (directory in workspace)243            commit: Commit hash or branch to reset to244 245        Returns:246            True if reset successful247 248        Raises:249            RuntimeError: If reset fails250        """251        repo_path = self.workspace_dir / repo_name252 253        if not repo_path.exists():254            raise RuntimeError(f"Repository not found in workspace: {repo_name}")255 256        # Fetch latest (in case commit is new)257        subprocess.run(258            ["git", "fetch", "--all"],259            cwd=str(repo_path),260            capture_output=True,261        )262 263        # Checkout and hard reset to commit264        result = subprocess.run(265            ["git", "checkout", commit],266            cwd=str(repo_path),267            capture_output=True,268            text=True,269        )270 271        if result.returncode != 0:272            raise RuntimeError(f"Checkout failed: {result.stderr}")273 274        result = subprocess.run(275            ["git", "reset", "--hard", f"origin/{commit}" if commit != "main" else commit],276            cwd=str(repo_path),277            capture_output=True,278            text=True,279        )280 281        if result.returncode != 0:282            # Try without origin/ prefix283            result = subprocess.run(284                ["git", "reset", "--hard", commit],285                cwd=str(repo_path),286                capture_output=True,287                text=True,288            )289            if result.returncode != 0:290                raise RuntimeError(f"Reset failed: {result.stderr}")291 292        # Clean untracked files and directories293        subprocess.run(294            ["git", "clean", "-fdx"],295            cwd=str(repo_path),296            capture_output=True,297        )298 299        return True300 301    def execute_git_command(302        self, command: str, working_dir: str = ""303    ) -> tuple[int, str, str]:304        """305        Execute a git command in the workspace.306 307        Args:308            command: Git command to execute (without 'git' prefix)309            working_dir: Working directory relative to workspace310 311        Returns:312            Tuple of (exit_code, stdout, stderr)313        """314        work_path = (315            self.workspace_dir / working_dir if working_dir else self.workspace_dir316        )317 318        if not work_path.exists():319            return (1, "", f"Working directory does not exist: {work_path}")320 321        # Split command safely322        cmd_parts = ["git"] + command.split()323 324        result = subprocess.run(325            cmd_parts,326            cwd=str(work_path),327            capture_output=True,328            text=True,329        )330 331        return (result.returncode, result.stdout, result.stderr)332 333    def get_current_commit(self, repo_name: str) -> str:334        """335        Get current commit hash of a workspace repository.336 337        Args:338            repo_name: Name of repository in workspace339 340        Returns:341            Commit hash342        """343        repo_path = self.workspace_dir / repo_name344 345        if not repo_path.exists():346            raise RuntimeError(f"Repository not found: {repo_name}")347 348        result = subprocess.run(349            ["git", "rev-parse", "HEAD"],350            cwd=str(repo_path),351            capture_output=True,352            text=True,353        )354 355        if result.returncode != 0:356            raise RuntimeError(f"Failed to get commit: {result.stderr}")357 358        return result.stdout.strip()359 360    def workspace_exists(self, repo_name: str) -> bool:361        """Check if a repository exists in workspace."""362        return (self.workspace_dir / repo_name).exists()363