diff --git a/.changeset/icy-ideas-burn.md b/.changeset/icy-ideas-burn.md new file mode 100644 index 0000000000..7711630c7a --- /dev/null +++ b/.changeset/icy-ideas-burn.md @@ -0,0 +1,5 @@ +--- +'@e2b/python-sdk': patch +--- + +Consolidate logic on the git functionality and tests to ensure parity on sync/async functionality diff --git a/packages/python-sdk/e2b/__init__.py b/packages/python-sdk/e2b/__init__.py index 4d842fb9c0..c0a9316120 100644 --- a/packages/python-sdk/e2b/__init__.py +++ b/packages/python-sdk/e2b/__init__.py @@ -60,7 +60,7 @@ FilesystemEvent, FilesystemEventType, ) -from .sandbox.git_utils import GitBranches, GitFileStatus, GitStatus +from .sandbox._git import GitBranches, GitFileStatus, GitStatus from .sandbox.network import ALL_TRAFFIC from .sandbox.sandbox_api import ( GitHubMcpServer, diff --git a/packages/python-sdk/e2b/sandbox/_git/__init__.py b/packages/python-sdk/e2b/sandbox/_git/__init__.py new file mode 100644 index 0000000000..8c005e0f62 --- /dev/null +++ b/packages/python-sdk/e2b/sandbox/_git/__init__.py @@ -0,0 +1,78 @@ +from e2b.sandbox._git.args import ( + build_add_args, + build_branches_args, + build_checkout_branch_args, + build_clone_plan, + build_commit_args, + build_credential_approve_command, + build_create_branch_args, + build_delete_branch_args, + build_git_command, + build_has_upstream_args, + build_pull_args, + build_push_args, + build_remote_add_args, + build_remote_add_shell_command, + build_remote_get_command, + build_remote_get_url_args, + build_remote_set_url_args, + build_reset_args, + build_restore_args, + build_status_args, + shell_escape, +) +from e2b.sandbox._git.auth import ( + build_auth_error_message, + build_upstream_error_message, + is_auth_failure, + is_missing_upstream, + strip_credentials, + with_credentials, +) +from e2b.sandbox._git.config import resolve_config_scope +from e2b.sandbox._git.parse import ( + derive_repo_dir_from_url, + parse_git_branches, + parse_git_status, + parse_remote_url, +) +from e2b.sandbox._git.types import ClonePlan, GitBranches, GitFileStatus, GitStatus + +__all__ = [ + "build_add_args", + "build_auth_error_message", + "build_branches_args", + "build_checkout_branch_args", + "build_clone_plan", + "build_commit_args", + "build_credential_approve_command", + "build_create_branch_args", + "build_delete_branch_args", + "build_git_command", + "build_has_upstream_args", + "build_pull_args", + "build_push_args", + "build_remote_add_args", + "build_remote_add_shell_command", + "build_remote_get_command", + "build_remote_get_url_args", + "build_remote_set_url_args", + "build_reset_args", + "build_restore_args", + "build_status_args", + "build_upstream_error_message", + "derive_repo_dir_from_url", + "is_auth_failure", + "is_missing_upstream", + "parse_git_branches", + "parse_git_status", + "parse_remote_url", + "resolve_config_scope", + "shell_escape", + "strip_credentials", + "with_credentials", + "ClonePlan", + "GitBranches", + "GitFileStatus", + "GitStatus", +] diff --git a/packages/python-sdk/e2b/sandbox/_git/args.py b/packages/python-sdk/e2b/sandbox/_git/args.py new file mode 100644 index 0000000000..0eeef13a0f --- /dev/null +++ b/packages/python-sdk/e2b/sandbox/_git/args.py @@ -0,0 +1,363 @@ +from typing import List, Optional + +from e2b.exceptions import InvalidArgumentException +from e2b.sandbox._git.auth import strip_credentials, with_credentials +from e2b.sandbox._git.parse import derive_repo_dir_from_url +from e2b.sandbox._git.types import ClonePlan + + +def shell_escape(value: str) -> str: + """ + Escape a string for safe use in a shell command. + + :param value: Value to escape + :return: Shell-escaped string + """ + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def build_git_command(args: List[str], repo_path: Optional[str] = None) -> str: + """ + Build a shell-safe git command string. + + :param args: Git command arguments + :param repo_path: Repository path for `git -C`, if provided + :return: Shell-safe git command + """ + parts = ["git"] + if repo_path: + parts.extend(["-C", repo_path]) + parts.extend(args) + return " ".join(shell_escape(part) for part in parts) + + +def build_push_args( + remote_name: Optional[str], + *, + remote: Optional[str], + branch: Optional[str], + set_upstream: bool, +) -> List[str]: + """ + Build arguments for a git push command. + + :param remote_name: Resolved remote name, if any + :param remote: Remote name override + :param branch: Branch name to push + :param set_upstream: Whether to set upstream tracking + :return: List of git push arguments + """ + args = ["push"] + target_remote = remote_name or remote + if set_upstream and target_remote: + args.append("--set-upstream") + if target_remote: + args.append(target_remote) + if branch: + args.append(branch) + return args + + +def build_pull_args( + remote: Optional[str], + branch: Optional[str], + remote_name: Optional[str] = None, +) -> List[str]: + """ + Build arguments for a git pull command. + + :param remote: Remote name override + :param branch: Branch name to pull + :param remote_name: Resolved remote name, if any + :return: List of git pull arguments + """ + args = ["pull"] + target_remote = remote_name or remote + if target_remote: + args.append(target_remote) + if branch: + args.append(branch) + return args + + +def build_remote_add_args(name: str, url: str, fetch: bool) -> List[str]: + """ + Build arguments for a git remote add command. + + :param name: Remote name + :param url: Remote URL + :param fetch: Whether to fetch after adding the remote + :return: List of git remote add arguments + """ + if not name or not url: + raise InvalidArgumentException( + "Both remote name and URL are required to add a git remote." + ) + + args = ["remote", "add"] + if fetch: + args.append("-f") + args.extend([name, url]) + return args + + +def build_remote_add_shell_command( + args: List[str], + path: str, + name: str, + url: str, + fetch: bool, +) -> str: + """ + Build a shell command that adds or updates a remote and optionally fetches. + + :param args: Base git remote add args + :param path: Repository path + :param name: Remote name + :param url: Remote URL + :param fetch: Whether to fetch after adding the remote + :return: Shell command string + """ + add_cmd = build_git_command(args, path) + set_url_cmd = build_git_command(build_remote_set_url_args(name, url), path) + cmd = f"{add_cmd} || {set_url_cmd}" + if fetch: + fetch_cmd = build_git_command(["fetch", name], path) + cmd = f"({cmd}) && {fetch_cmd}" + return cmd + + +def build_remote_get_url_args(name: str) -> List[str]: + """ + Build arguments for a git remote get-url command. + """ + return ["remote", "get-url", name] + + +def build_remote_set_url_args(name: str, url: str) -> List[str]: + """ + Build arguments for a git remote set-url command. + """ + return ["remote", "set-url", name, url] + + +def build_remote_get_command(path: str, name: str) -> str: + """ + Build a shell command that returns the remote URL or empty output. + + :param path: Repository path + :param name: Remote name + :return: Shell command string + """ + if not name: + raise InvalidArgumentException("Remote name is required.") + + return f"{build_git_command(build_remote_get_url_args(name), path)} || true" + + +def build_credential_approve_command( + username: str, + password: str, + host: str, + protocol: str, +) -> str: + """ + Build a git credential approve command for the given credentials. + """ + target_host = host.strip() or "github.com" + target_protocol = protocol.strip() or "https" + credential_input = "\n".join( + [ + f"protocol={target_protocol}", + f"host={target_host}", + f"username={username}", + f"password={password}", + "", + "", + ] + ) + return ( + f"printf %s {shell_escape(credential_input)} | " + f"{build_git_command(['credential', 'approve'])}" + ) + + +def build_has_upstream_args() -> List[str]: + """ + Build arguments for a git upstream check command. + """ + return ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"] + + +def build_status_args() -> List[str]: + """ + Build arguments for a git status command. + """ + return ["status", "--porcelain=1", "-b"] + + +def build_branches_args() -> List[str]: + """ + Build arguments for a git branch listing command. + """ + return ["branch", "--format=%(refname:short)\t%(HEAD)"] + + +def build_create_branch_args(branch: str) -> List[str]: + """ + Build arguments for a git checkout -b command. + """ + return ["checkout", "-b", branch] + + +def build_checkout_branch_args(branch: str) -> List[str]: + """ + Build arguments for a git checkout command. + """ + return ["checkout", branch] + + +def build_delete_branch_args(branch: str, force: bool) -> List[str]: + """ + Build arguments for a git branch delete command. + """ + return ["branch", "-D" if force else "-d", branch] + + +def build_add_args(files: Optional[List[str]], all: bool) -> List[str]: + """ + Build arguments for a git add command. + """ + args = ["add"] + if not files: + args.append("-A" if all else ".") + else: + args.append("--") + args.extend(files) + return args + + +def build_commit_args( + message: str, + author_name: Optional[str], + author_email: Optional[str], + allow_empty: bool, +) -> List[str]: + """ + Build arguments for a git commit command. + """ + args = ["commit", "-m", message] + if allow_empty: + args.append("--allow-empty") + author_args: List[str] = [] + if author_name: + author_args.extend(["-c", f"user.name={author_name}"]) + if author_email: + author_args.extend(["-c", f"user.email={author_email}"]) + if author_args: + args = author_args + args + return args + + +def build_reset_args( + mode: Optional[str], + target: Optional[str], + paths: Optional[List[str]], +) -> List[str]: + """ + Build arguments for a git reset command. + """ + allowed_modes = {"soft", "mixed", "hard", "merge", "keep"} + if mode and mode not in allowed_modes: + raise InvalidArgumentException( + f"Reset mode must be one of {', '.join(sorted(allowed_modes))}." + ) + + args = ["reset"] + if mode: + args.append(f"--{mode}") + if target: + args.append(target) + if paths: + args.append("--") + args.extend(paths) + return args + + +def build_restore_args( + paths: List[str], + staged: Optional[bool], + worktree: Optional[bool], + source: Optional[str], +) -> List[str]: + """ + Build arguments for a git restore command. + """ + if not paths: + raise InvalidArgumentException("At least one path is required.") + + resolved_staged = staged + resolved_worktree = worktree + if staged is None and worktree is None: + resolved_worktree = True + elif staged is True and worktree is None: + resolved_worktree = False + elif staged is None and worktree is not None: + resolved_staged = False + + if resolved_staged is False and resolved_worktree is False: + raise InvalidArgumentException( + "At least one of staged or worktree must be true." + ) + + args = ["restore"] + if resolved_worktree: + args.append("--worktree") + if resolved_staged: + args.append("--staged") + if source: + args.extend(["--source", source]) + args.append("--") + args.extend(paths) + return args + + +def build_clone_plan( + url: str, + path: Optional[str], + branch: Optional[str], + depth: Optional[int], + auth_username: Optional[str], + auth_password: Optional[str], + dangerously_store_credentials: bool, +) -> ClonePlan: + """ + Build clone arguments and metadata for post-clone credential stripping. + """ + clone_url = ( + with_credentials(url, auth_username, auth_password) + if auth_username and auth_password + else url + ) + sanitized_url = strip_credentials(clone_url) + should_strip = not dangerously_store_credentials and sanitized_url != clone_url + repo_path = path if not should_strip else path or derive_repo_dir_from_url(url) + if should_strip and not repo_path: + raise InvalidArgumentException( + "A destination path is required when using credentials without storing them." + ) + + args = ["clone", clone_url] + if branch: + args.extend(["--branch", branch, "--single-branch"]) + if depth: + args.extend(["--depth", str(depth)]) + if path: + args.append(path) + + return ClonePlan( + args=args, + repo_path=repo_path, + sanitized_url=sanitized_url if should_strip else None, + should_strip=should_strip, + ) diff --git a/packages/python-sdk/e2b/sandbox/_git/auth.py b/packages/python-sdk/e2b/sandbox/_git/auth.py new file mode 100644 index 0000000000..a93511f48d --- /dev/null +++ b/packages/python-sdk/e2b/sandbox/_git/auth.py @@ -0,0 +1,132 @@ +from typing import Optional +from urllib.parse import urlparse, urlunparse + +from e2b.exceptions import InvalidArgumentException +from e2b.sandbox.commands.command_handle import CommandExitException + + +def with_credentials(url: str, username: Optional[str], password: Optional[str]) -> str: + """ + Add HTTP(S) credentials to a Git URL. + + :param url: Git repository URL + :param username: Username for HTTP(S) authentication + :param password: Password or token for HTTP(S) authentication + :return: URL with embedded credentials + """ + if not username and not password: + return url + if not username or not password: + raise InvalidArgumentException( + "Both username and password are required when using Git credentials." + ) + + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + raise InvalidArgumentException( + "Only http(s) Git URLs support username/password credentials." + ) + + netloc = f"{username}:{password}@{parsed.netloc}" + return urlunparse(parsed._replace(netloc=netloc)) + + +def strip_credentials(url: str) -> str: + """ + Strip HTTP(S) credentials from a Git URL. + + :param url: Git repository URL + :return: URL without embedded credentials + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + return url + if not parsed.username and not parsed.password: + return url + + host = parsed.hostname or "" + if parsed.port: + host = f"{host}:{parsed.port}" + + return urlunparse(parsed._replace(netloc=host)) + + +def is_auth_failure(err: Exception) -> bool: + """ + Check whether a git command failed due to authentication issues. + + :param err: Exception raised by a git command + :return: True when the error matches common authentication failures + """ + if not isinstance(err, CommandExitException): + return False + + message = f"{err.stderr}\n{err.stdout}".lower() + auth_snippets = [ + "authentication failed", + "terminal prompts disabled", + "could not read username", + "invalid username or password", + "access denied", + "permission denied", + "not authorized", + ] + return any(snippet in message for snippet in auth_snippets) + + +def is_missing_upstream(err: Exception) -> bool: + """ + Check whether a git command failed due to missing upstream tracking. + + :param err: Exception raised by a git command + :return: True when the error matches common upstream failures + """ + if not isinstance(err, CommandExitException): + return False + + message = f"{err.stderr}\n{err.stdout}".lower() + upstream_snippets = [ + "has no upstream branch", + "no upstream branch", + "no upstream configured", + "no tracking information for the current branch", + "no tracking information", + "set the remote as upstream", + "set the upstream branch", + "please specify which branch you want to merge with", + ] + return any(snippet in message for snippet in upstream_snippets) + + +def build_auth_error_message(action: str, missing_password: bool) -> str: + """ + Build a git authentication error message for the given action. + + :param action: Git action name + :param missing_password: Whether the password/token is missing + :return: Error message string + """ + if missing_password: + return f"Git {action} requires a password/token for private repositories." + return f"Git {action} requires credentials for private repositories." + + +def build_upstream_error_message(action: str) -> str: + """ + Build a git upstream tracking error message for the given action. + + :param action: Git action name + :return: Error message string + """ + if action == "push": + return ( + "Git push failed because no upstream branch is configured. " + "Set upstream once with set_upstream=True (and optional remote/branch), " + "or pass remote and branch explicitly." + ) + + return ( + "Git pull failed because no upstream branch is configured. " + "Pass remote and branch explicitly, or set upstream once (push with " + "set_upstream=True or run: git branch --set-upstream-to=origin/ )." + ) diff --git a/packages/python-sdk/e2b/sandbox/_git/config.py b/packages/python-sdk/e2b/sandbox/_git/config.py new file mode 100644 index 0000000000..0c851b046f --- /dev/null +++ b/packages/python-sdk/e2b/sandbox/_git/config.py @@ -0,0 +1,32 @@ +from typing import Optional + +from e2b.exceptions import InvalidArgumentException + + +def resolve_config_scope( + scope: Optional[str], path: Optional[str] +) -> tuple[str, Optional[str]]: + """ + Resolve a git config scope flag and repository path. + + :param scope: Requested scope ("global", "local", "system") + :param path: Repository path for local scope + :return: Tuple of (scope flag, repository path) + """ + scope_name = (scope or "global").strip().lower() + if scope_name not in {"global", "local", "system"}: + raise InvalidArgumentException( + "Git config scope must be one of: global, local, system." + ) + + if scope_name == "local": + if not path: + raise InvalidArgumentException( + "Repository path is required when scope is local." + ) + return "--local", path + + if scope_name == "system": + return "--system", None + + return "--global", None diff --git a/packages/python-sdk/e2b/sandbox/_git/parse.py b/packages/python-sdk/e2b/sandbox/_git/parse.py new file mode 100644 index 0000000000..e6cea693af --- /dev/null +++ b/packages/python-sdk/e2b/sandbox/_git/parse.py @@ -0,0 +1,222 @@ +from typing import List, Optional +from urllib.parse import urlparse + +from e2b.exceptions import InvalidArgumentException +from e2b.sandbox._git.types import GitBranches, GitFileStatus, GitStatus + + +def derive_repo_dir_from_url(url: str) -> Optional[str]: + """ + Derive the default repository directory name from a Git URL. + + :param url: Git repository URL + :return: Repository directory name, if it can be determined + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + return None + trimmed_path = parsed.path.rstrip("/") + if not trimmed_path: + return None + last_segment = trimmed_path.split("/")[-1] + if not last_segment: + return None + return last_segment[:-4] if last_segment.endswith(".git") else last_segment + + +def _parse_ahead_behind(segment: Optional[str]) -> tuple[int, int]: + """ + Parse the ahead/behind segment from porcelain branch info. + + :param segment: Segment text like "ahead 2, behind 1" + :return: Tuple of (ahead, behind) + """ + if not segment: + return 0, 0 + ahead = 0 + behind = 0 + if "ahead" in segment: + try: + ahead = int(segment.split("ahead")[1].split(",")[0].strip()) + except Exception: + ahead = 0 + if "behind" in segment: + try: + behind = int(segment.split("behind")[1].split(",")[0].strip()) + except Exception: + behind = 0 + return ahead, behind + + +def _normalize_branch_name(name: str) -> str: + """ + Normalize branch names from porcelain branch output. + + :param name: Raw branch name section + :return: Normalized branch name + """ + if name.startswith("HEAD (detached at "): + return name.replace("HEAD (detached at ", "").rstrip(")") + return ( + name.replace("HEAD (no branch)", "HEAD") + .replace("No commits yet on ", "") + .replace("Initial commit on ", "") + ) + + +def _derive_status(index_status: str, working_status: str) -> str: + """ + Derive a normalized status label from porcelain status characters. + + :param index_status: Index status character + :param working_status: Working tree status character + :return: Normalized status label + """ + statuses = {index_status, working_status} + if "U" in statuses: + return "conflict" + if "R" in statuses: + return "renamed" + if "C" in statuses: + return "copied" + if "D" in statuses: + return "deleted" + if "A" in statuses: + return "added" + if "M" in statuses: + return "modified" + if "T" in statuses: + return "typechange" + if "?" in statuses: + return "untracked" + return "unknown" + + +def parse_git_status(output: str) -> GitStatus: + """ + Parse `git status --porcelain=1 -b` output into a structured object. + + :param output: Git status output + :return: Parsed `GitStatus` + """ + lines = [line.rstrip() for line in output.split("\n") if line.strip()] + current_branch: Optional[str] = None + upstream: Optional[str] = None + ahead = 0 + behind = 0 + detached = False + file_status: List[GitFileStatus] = [] + + if not lines: + return GitStatus( + current_branch=current_branch, + upstream=upstream, + ahead=ahead, + behind=behind, + detached=detached, + file_status=file_status, + ) + + branch_line = lines[0] + if branch_line.startswith("## "): + branch_info = branch_line[3:] + ahead_start = branch_info.find(" [") + branch_part = branch_info if ahead_start == -1 else branch_info[:ahead_start] + ahead_part = None if ahead_start == -1 else branch_info[ahead_start + 2 : -1] + normalized_branch = _normalize_branch_name(branch_part) + raw_branch = branch_part + is_detached = raw_branch.startswith("HEAD (detached at ") or ( + "detached" in raw_branch + ) + + if is_detached or normalized_branch.startswith("HEAD"): + detached = True + elif "..." in normalized_branch: + branch, upstream_branch = normalized_branch.split("...") + current_branch = branch or None + upstream = upstream_branch or None + else: + current_branch = normalized_branch or None + + ahead, behind = _parse_ahead_behind(ahead_part) + + for line in lines[1:]: + if line.startswith("?? "): + name = line[3:] + file_status.append( + GitFileStatus( + name=name, + status="untracked", + index_status="?", + working_tree_status="?", + staged=False, + ) + ) + continue + + if len(line) < 3: + continue + index_status = line[0] + working_status = line[1] + path = line[3:] + renamed_from: Optional[str] = None + name = path + if " -> " in path: + renamed_from, name = path.split(" -> ", 1) + + file_status.append( + GitFileStatus( + name=name, + status=_derive_status(index_status, working_status), + index_status=index_status, + working_tree_status=working_status, + staged=index_status not in (" ", "?"), + renamed_from=renamed_from, + ) + ) + + return GitStatus( + current_branch=current_branch, + upstream=upstream, + ahead=ahead, + behind=behind, + detached=detached, + file_status=file_status, + ) + + +def parse_git_branches(output: str) -> GitBranches: + """ + Parse `git branch --format=%(refname:short)\t%(HEAD)` output. + + :param output: Git branch output + :return: Parsed `GitBranches` + """ + branches: List[str] = [] + current_branch: Optional[str] = None + + lines = [line.strip() for line in output.split("\n") if line.strip()] + for line in lines: + parts = line.split("\t") + name = parts[0] + branches.append(name) + if len(parts) > 1 and parts[1] == "*": + current_branch = name + + return GitBranches(branches=branches, current_branch=current_branch) + + +def parse_remote_url(output: str, remote: str) -> str: + """ + Parse a git remote URL output and validate it's present. + + :param output: Git remote get-url output + :param remote: Remote name for the error message + :return: Remote URL + """ + url = output.strip() + if not url: + raise InvalidArgumentException( + f'Remote "{remote}" URL not found in repository.' + ) + return url diff --git a/packages/python-sdk/e2b/sandbox/_git/types.py b/packages/python-sdk/e2b/sandbox/_git/types.py new file mode 100644 index 0000000000..2ddb70f294 --- /dev/null +++ b/packages/python-sdk/e2b/sandbox/_git/types.py @@ -0,0 +1,144 @@ +from dataclasses import dataclass +from typing import List, Optional + + +@dataclass +class GitFileStatus: + """ + Parsed git status entry for a file. + + :param name: Path relative to the repository root + :param status: Normalized status string (e.g. "modified", "added") + :param index_status: Index status character from porcelain output + :param working_tree_status: Working tree status character from porcelain output + :param staged: Whether the change is staged + :param renamed_from: Original path when the file was renamed + """ + + name: str + status: str + index_status: str + working_tree_status: str + staged: bool + renamed_from: Optional[str] = None + + +@dataclass +class GitStatus: + """ + Parsed git repository status. + + :param current_branch: Current branch name, if available + :param upstream: Upstream branch name, if available + :param ahead: Number of commits the branch is ahead of upstream + :param behind: Number of commits the branch is behind upstream + :param detached: Whether HEAD is detached + :param file_status: List of file status entries + """ + + current_branch: Optional[str] + upstream: Optional[str] + ahead: int + behind: int + detached: bool + file_status: List[GitFileStatus] + + @property + def is_clean(self) -> bool: + """ + Return True when there are no tracked or untracked file changes. + """ + return len(self.file_status) == 0 + + @property + def has_changes(self) -> bool: + """ + Return True when there are any tracked or untracked file changes. + """ + return len(self.file_status) > 0 + + @property + def has_staged(self) -> bool: + """ + Return True when at least one file has staged changes. + """ + return any(item.staged for item in self.file_status) + + @property + def has_untracked(self) -> bool: + """ + Return True when at least one file is untracked. + """ + return any(item.status == "untracked" for item in self.file_status) + + @property + def has_conflicts(self) -> bool: + """ + Return True when at least one file is in conflict. + """ + return any(item.status == "conflict" for item in self.file_status) + + @property + def total_count(self) -> int: + """ + Return the total number of changed files. + """ + return len(self.file_status) + + @property + def staged_count(self) -> int: + """ + Return the number of files with staged changes. + """ + return sum(1 for item in self.file_status if item.staged) + + @property + def unstaged_count(self) -> int: + """ + Return the number of files with unstaged changes. + """ + return sum(1 for item in self.file_status if not item.staged) + + @property + def untracked_count(self) -> int: + """ + Return the number of untracked files. + """ + return sum(1 for item in self.file_status if item.status == "untracked") + + @property + def conflict_count(self) -> int: + """ + Return the number of files with merge conflicts. + """ + return sum(1 for item in self.file_status if item.status == "conflict") + + +@dataclass +class GitBranches: + """ + Parsed git branch list. + + :param branches: List of branch names + :param current_branch: Current branch name, if available + """ + + branches: List[str] + current_branch: Optional[str] + + +@dataclass +class ClonePlan: + """ + Prepared arguments and metadata for git clone. + + :param args: Command arguments for git clone + :param repo_path: Repository path to use for post-clone adjustments + :param sanitized_url: Credential-stripped URL to restore + :param should_strip: Whether to reset the remote URL after clone + """ + + args: List[str] + repo_path: Optional[str] + sanitized_url: Optional[str] + should_strip: bool diff --git a/packages/python-sdk/e2b/sandbox/git_utils.py b/packages/python-sdk/e2b/sandbox/git_utils.py deleted file mode 100644 index 3ac044d486..0000000000 --- a/packages/python-sdk/e2b/sandbox/git_utils.py +++ /dev/null @@ -1,540 +0,0 @@ -from dataclasses import dataclass -from typing import List, Optional -from urllib.parse import urlparse, urlunparse - -from e2b.exceptions import InvalidArgumentException -from e2b.sandbox.commands.command_handle import CommandExitException - - -@dataclass -class GitFileStatus: - """ - Parsed git status entry for a file. - - :param name: Path relative to the repository root - :param status: Normalized status string (e.g. "modified", "added") - :param index_status: Index status character from porcelain output - :param working_tree_status: Working tree status character from porcelain output - :param staged: Whether the change is staged - :param renamed_from: Original path when the file was renamed - """ - - name: str - status: str - index_status: str - working_tree_status: str - staged: bool - renamed_from: Optional[str] = None - - -@dataclass -class GitStatus: - """ - Parsed git repository status. - - :param current_branch: Current branch name, if available - :param upstream: Upstream branch name, if available - :param ahead: Number of commits the branch is ahead of upstream - :param behind: Number of commits the branch is behind upstream - :param detached: Whether HEAD is detached - :param file_status: List of file status entries - """ - - current_branch: Optional[str] - upstream: Optional[str] - ahead: int - behind: int - detached: bool - file_status: List[GitFileStatus] - - @property - def is_clean(self) -> bool: - """ - Return True when there are no tracked or untracked file changes. - """ - return len(self.file_status) == 0 - - @property - def has_changes(self) -> bool: - """ - Return True when there are any tracked or untracked file changes. - """ - return len(self.file_status) > 0 - - @property - def has_staged(self) -> bool: - """ - Return True when at least one file has staged changes. - """ - return any(item.staged for item in self.file_status) - - @property - def has_untracked(self) -> bool: - """ - Return True when at least one file is untracked. - """ - return any(item.status == "untracked" for item in self.file_status) - - @property - def has_conflicts(self) -> bool: - """ - Return True when at least one file is in conflict. - """ - return any(item.status == "conflict" for item in self.file_status) - - @property - def total_count(self) -> int: - """ - Return the total number of changed files. - """ - return len(self.file_status) - - @property - def staged_count(self) -> int: - """ - Return the number of files with staged changes. - """ - return sum(1 for item in self.file_status if item.staged) - - @property - def unstaged_count(self) -> int: - """ - Return the number of files with unstaged changes. - """ - return sum(1 for item in self.file_status if not item.staged) - - @property - def untracked_count(self) -> int: - """ - Return the number of untracked files. - """ - return sum(1 for item in self.file_status if item.status == "untracked") - - @property - def conflict_count(self) -> int: - """ - Return the number of files with merge conflicts. - """ - return sum(1 for item in self.file_status if item.status == "conflict") - - -@dataclass -class GitBranches: - """ - Parsed git branch list. - - :param branches: List of branch names - :param current_branch: Current branch name, if available - """ - - branches: List[str] - current_branch: Optional[str] - - -def shell_escape(value: str) -> str: - """ - Escape a string for safe use in a shell command. - - :param value: Value to escape - :return: Shell-escaped string - """ - return "'" + value.replace("'", "'\"'\"'") + "'" - - -def with_credentials(url: str, username: Optional[str], password: Optional[str]) -> str: - """ - Add HTTP(S) credentials to a Git URL. - - :param url: Git repository URL - :param username: Username for HTTP(S) authentication - :param password: Password or token for HTTP(S) authentication - :return: URL with embedded credentials - """ - if not username and not password: - return url - if not username or not password: - raise InvalidArgumentException( - "Both username and password are required when using Git credentials." - ) - - parsed = urlparse(url) - if parsed.scheme not in ("http", "https"): - raise InvalidArgumentException( - "Only http(s) Git URLs support username/password credentials." - ) - - netloc = f"{username}:{password}@{parsed.netloc}" - return urlunparse(parsed._replace(netloc=netloc)) - - -def strip_credentials(url: str) -> str: - """ - Strip HTTP(S) credentials from a Git URL. - - :param url: Git repository URL - :return: URL without embedded credentials - """ - parsed = urlparse(url) - if parsed.scheme not in ("http", "https"): - return url - if not parsed.username and not parsed.password: - return url - - host = parsed.hostname or "" - if parsed.port: - host = f"{host}:{parsed.port}" - - return urlunparse(parsed._replace(netloc=host)) - - -def derive_repo_dir_from_url(url: str) -> Optional[str]: - """ - Derive the default repository directory name from a Git URL. - - :param url: Git repository URL - :return: Repository directory name, if it can be determined - """ - parsed = urlparse(url) - if parsed.scheme not in ("http", "https"): - return None - trimmed_path = parsed.path.rstrip("/") - if not trimmed_path: - return None - last_segment = trimmed_path.split("/")[-1] - if not last_segment: - return None - return last_segment[:-4] if last_segment.endswith(".git") else last_segment - - -def build_git_command(args: List[str], repo_path: Optional[str] = None) -> str: - """ - Build a shell-safe git command string. - - :param args: Git command arguments - :param repo_path: Repository path for `git -C`, if provided - :return: Shell-safe git command - """ - parts = ["git"] - if repo_path: - parts.extend(["-C", repo_path]) - parts.extend(args) - return " ".join(shell_escape(part) for part in parts) - - -def build_push_args( - remote_name: Optional[str], - *, - remote: Optional[str], - branch: Optional[str], - set_upstream: bool, -) -> List[str]: - """ - Build arguments for a git push command. - - :param remote_name: Resolved remote name, if any - :param remote: Remote name override - :param branch: Branch name to push - :param set_upstream: Whether to set upstream tracking - :return: List of git push arguments - """ - args = ["push"] - target_remote = remote_name or remote - if set_upstream and target_remote: - args.append("--set-upstream") - if target_remote: - args.append(target_remote) - if branch: - args.append(branch) - return args - - -def is_auth_failure(err: Exception) -> bool: - """ - Check whether a git command failed due to authentication issues. - - :param err: Exception raised by a git command - :return: True when the error matches common authentication failures - """ - if not isinstance(err, CommandExitException): - return False - - message = f"{err.stderr}\n{err.stdout}".lower() - auth_snippets = [ - "authentication failed", - "terminal prompts disabled", - "could not read username", - "invalid username or password", - "access denied", - "permission denied", - "not authorized", - ] - return any(snippet in message for snippet in auth_snippets) - - -def is_missing_upstream(err: Exception) -> bool: - """ - Check whether a git command failed due to missing upstream tracking. - - :param err: Exception raised by a git command - :return: True when the error matches common upstream failures - """ - if not isinstance(err, CommandExitException): - return False - - message = f"{err.stderr}\n{err.stdout}".lower() - upstream_snippets = [ - "has no upstream branch", - "no upstream branch", - "no upstream configured", - "no tracking information for the current branch", - "no tracking information", - "set the remote as upstream", - "set the upstream branch", - "please specify which branch you want to merge with", - ] - return any(snippet in message for snippet in upstream_snippets) - - -def build_auth_error_message(action: str, missing_password: bool) -> str: - """ - Build a git authentication error message for the given action. - - :param action: Git action name - :param missing_password: Whether the password/token is missing - :return: Error message string - """ - if missing_password: - return f"Git {action} requires a password/token for private repositories." - return f"Git {action} requires credentials for private repositories." - - -def build_upstream_error_message(action: str) -> str: - """ - Build a git upstream tracking error message for the given action. - - :param action: Git action name - :return: Error message string - """ - if action == "push": - return ( - "Git push failed because no upstream branch is configured. " - "Set upstream once with set_upstream=True (and optional remote/branch), " - "or pass remote and branch explicitly." - ) - - return ( - "Git pull failed because no upstream branch is configured. " - "Pass remote and branch explicitly, or set upstream once (push with " - "set_upstream=True or run: git branch --set-upstream-to=origin/ )." - ) - - -def resolve_config_scope( - scope: Optional[str], path: Optional[str] -) -> tuple[str, Optional[str]]: - """ - Resolve a git config scope flag and repository path. - - :param scope: Requested scope ("global", "local", "system") - :param path: Repository path for local scope - :return: Tuple of (scope flag, repository path) - """ - scope_name = (scope or "global").strip().lower() - if scope_name not in {"global", "local", "system"}: - raise InvalidArgumentException( - "Git config scope must be one of: global, local, system." - ) - - if scope_name == "local": - if not path: - raise InvalidArgumentException( - "Repository path is required when scope is local." - ) - return "--local", path - - if scope_name == "system": - return "--system", None - - return "--global", None - - -def _parse_ahead_behind(segment: Optional[str]) -> tuple[int, int]: - """ - Parse the ahead/behind segment from porcelain branch info. - - :param segment: Segment text like "ahead 2, behind 1" - :return: Tuple of (ahead, behind) - """ - if not segment: - return 0, 0 - ahead = 0 - behind = 0 - if "ahead" in segment: - try: - ahead = int(segment.split("ahead")[1].split(",")[0].strip()) - except Exception: - ahead = 0 - if "behind" in segment: - try: - behind = int(segment.split("behind")[1].split(",")[0].strip()) - except Exception: - behind = 0 - return ahead, behind - - -def _normalize_branch_name(name: str) -> str: - """ - Normalize branch names from porcelain branch output. - - :param name: Raw branch name section - :return: Normalized branch name - """ - if name.startswith("HEAD (detached at "): - return name.replace("HEAD (detached at ", "").rstrip(")") - return ( - name.replace("HEAD (no branch)", "HEAD") - .replace("No commits yet on ", "") - .replace("Initial commit on ", "") - ) - - -def _derive_status(index_status: str, working_status: str) -> str: - """ - Derive a normalized status label from porcelain status characters. - - :param index_status: Index status character - :param working_status: Working tree status character - :return: Normalized status label - """ - statuses = {index_status, working_status} - if "U" in statuses: - return "conflict" - if "R" in statuses: - return "renamed" - if "C" in statuses: - return "copied" - if "D" in statuses: - return "deleted" - if "A" in statuses: - return "added" - if "M" in statuses: - return "modified" - if "T" in statuses: - return "typechange" - if "?" in statuses: - return "untracked" - return "unknown" - - -def parse_git_status(output: str) -> GitStatus: - """ - Parse `git status --porcelain=1 -b` output into a structured object. - - :param output: Git status output - :return: Parsed `GitStatus` - """ - lines = [line.rstrip() for line in output.split("\n") if line.strip()] - current_branch: Optional[str] = None - upstream: Optional[str] = None - ahead = 0 - behind = 0 - detached = False - file_status: List[GitFileStatus] = [] - - if not lines: - return GitStatus( - current_branch=current_branch, - upstream=upstream, - ahead=ahead, - behind=behind, - detached=detached, - file_status=file_status, - ) - - branch_line = lines[0] - if branch_line.startswith("## "): - branch_info = branch_line[3:] - ahead_start = branch_info.find(" [") - branch_part = branch_info if ahead_start == -1 else branch_info[:ahead_start] - ahead_part = None if ahead_start == -1 else branch_info[ahead_start + 2 : -1] - normalized_branch = _normalize_branch_name(branch_part) - raw_branch = branch_part - is_detached = raw_branch.startswith("HEAD (detached at ") or ( - "detached" in raw_branch - ) - - if is_detached or normalized_branch.startswith("HEAD"): - detached = True - elif "..." in normalized_branch: - branch, upstream_branch = normalized_branch.split("...") - current_branch = branch or None - upstream = upstream_branch or None - else: - current_branch = normalized_branch or None - - ahead, behind = _parse_ahead_behind(ahead_part) - - for line in lines[1:]: - if line.startswith("?? "): - name = line[3:] - file_status.append( - GitFileStatus( - name=name, - status="untracked", - index_status="?", - working_tree_status="?", - staged=False, - ) - ) - continue - - if len(line) < 3: - continue - index_status = line[0] - working_status = line[1] - path = line[3:] - renamed_from: Optional[str] = None - name = path - if " -> " in path: - renamed_from, name = path.split(" -> ", 1) - - file_status.append( - GitFileStatus( - name=name, - status=_derive_status(index_status, working_status), - index_status=index_status, - working_tree_status=working_status, - staged=index_status not in (" ", "?"), - renamed_from=renamed_from, - ) - ) - - return GitStatus( - current_branch=current_branch, - upstream=upstream, - ahead=ahead, - behind=behind, - detached=detached, - file_status=file_status, - ) - - -def parse_git_branches(output: str) -> GitBranches: - """ - Parse `git branch --format=%(refname:short)\t%(HEAD)` output. - - :param output: Git branch output - :return: Parsed `GitBranches` - """ - branches: List[str] = [] - current_branch: Optional[str] = None - - lines = [line.strip() for line in output.split("\n") if line.strip()] - for line in lines: - parts = line.split("\t") - name = parts[0] - branches.append(name) - if len(parts) > 1 and parts[1] == "*": - current_branch = name - - return GitBranches(branches=branches, current_branch=current_branch) diff --git a/packages/python-sdk/e2b/sandbox_async/git.py b/packages/python-sdk/e2b/sandbox_async/git.py index 1a2d194547..9b9f3a686e 100644 --- a/packages/python-sdk/e2b/sandbox_async/git.py +++ b/packages/python-sdk/e2b/sandbox_async/git.py @@ -6,21 +6,37 @@ InvalidArgumentException, ) from e2b.sandbox.commands.command_handle import CommandExitException -from e2b.sandbox.git_utils import ( +from e2b.sandbox._git import ( GitBranches, GitStatus, + build_add_args, build_auth_error_message, + build_branches_args, + build_checkout_branch_args, + build_clone_plan, + build_commit_args, + build_credential_approve_command, + build_create_branch_args, + build_delete_branch_args, build_git_command, + build_has_upstream_args, + build_pull_args, build_push_args, + build_remote_add_args, + build_remote_add_shell_command, + build_remote_get_command, + build_remote_get_url_args, + build_remote_set_url_args, + build_reset_args, + build_restore_args, + build_status_args, build_upstream_error_message, is_auth_failure, is_missing_upstream, parse_git_branches, parse_git_status, + parse_remote_url, resolve_config_scope, - shell_escape, - strip_credentials, - derive_repo_dir_from_url, with_credentials, ) from e2b.sandbox_async.commands.command import Commands @@ -116,7 +132,7 @@ async def _has_upstream( ) -> bool: try: result = await self._run_git( - ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"], + build_has_upstream_args(), path, envs, user, @@ -226,7 +242,7 @@ async def _get_remote_url( request_timeout: Optional[float] = None, ) -> str: result = await self._run_git( - ["remote", "get-url", remote], + build_remote_get_url_args(remote), path, envs, user, @@ -234,12 +250,7 @@ async def _get_remote_url( timeout, request_timeout, ) - url = result.stdout.strip() - if not url: - raise InvalidArgumentException( - f'Remote "{remote}" URL not found in repository.' - ) - return url + return parse_remote_url(result.stdout, remote) async def clone( self, @@ -281,36 +292,22 @@ async def clone( async def attempt_clone( auth_username: Optional[str], auth_password: Optional[str] ): - clone_url = ( - with_credentials(url, auth_username, auth_password) - if auth_username and auth_password - else url - ) - sanitized_url = strip_credentials(clone_url) - should_strip = ( - not dangerously_store_credentials and sanitized_url != clone_url - ) - repo_path = ( - path if not should_strip else path or derive_repo_dir_from_url(url) + plan = build_clone_plan( + url=url, + path=path, + branch=branch, + depth=depth, + auth_username=auth_username, + auth_password=auth_password, + dangerously_store_credentials=dangerously_store_credentials, ) - if should_strip and not repo_path: - raise InvalidArgumentException( - "A destination path is required when using credentials without storing them." - ) - args = ["clone", clone_url] - if branch: - args.extend(["--branch", branch, "--single-branch"]) - if depth: - args.extend(["--depth", str(depth)]) - if path: - args.append(path) result = await self._run_git( - args, None, envs, user, cwd, timeout, request_timeout + plan.args, None, envs, user, cwd, timeout, request_timeout ) - if should_strip and repo_path: + if plan.should_strip and plan.repo_path and plan.sanitized_url: await self._run_git( - ["remote", "set-url", "origin", sanitized_url], - repo_path, + build_remote_set_url_args("origin", plan.sanitized_url), + plan.repo_path, envs, user, cwd, @@ -390,27 +387,14 @@ async def remote_add( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - if not name or not url: - raise InvalidArgumentException( - "Both remote name and URL are required to add a git remote." - ) - - args = ["remote", "add"] - if fetch: - args.append("-f") - args.extend([name, url]) + args = build_remote_add_args(name, url, fetch) if not overwrite: return await self._run_git( args, path, envs, user, cwd, timeout, request_timeout ) - add_cmd = build_git_command(args, path) - set_url_cmd = build_git_command(["remote", "set-url", name, url], path) - cmd = f"{add_cmd} || {set_url_cmd}" - if fetch: - fetch_cmd = build_git_command(["fetch", name], path) - cmd = f"({cmd}) && {fetch_cmd}" + cmd = build_remote_add_shell_command(args, path, name, url, fetch) return await self._run_shell( cmd, envs, @@ -444,10 +428,7 @@ async def remote_get( :param request_timeout: Timeout for the request in **seconds** :return: Remote URL if present, otherwise `None` """ - if not name: - raise InvalidArgumentException("Remote name is required.") - - cmd = f"{build_git_command(['remote', 'get-url', name], path)} || true" + cmd = build_remote_get_command(path, name) result = ( await self._run_shell( cmd, @@ -481,7 +462,7 @@ async def status( :return: Parsed git status """ result = await self._run_git( - ["status", "--porcelain=1", "-b"], + build_status_args(), path, envs, user, @@ -512,7 +493,7 @@ async def branches( :return: Parsed branch list """ result = await self._run_git( - ["branch", "--format=%(refname:short)\t%(HEAD)"], + build_branches_args(), path, envs, user, @@ -545,7 +526,7 @@ async def create_branch( :return: Command result from the command runner """ return await self._run_git( - ["checkout", "-b", branch], + build_create_branch_args(branch), path, envs, user, @@ -577,7 +558,7 @@ async def checkout_branch( :return: Command result from the command runner """ return await self._run_git( - ["checkout", branch], + build_checkout_branch_args(branch), path, envs, user, @@ -610,7 +591,7 @@ async def delete_branch( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - args = ["branch", "-D" if force else "-d", branch] + args = build_delete_branch_args(branch, force) return await self._run_git( args, path, envs, user, cwd, timeout, request_timeout ) @@ -639,12 +620,7 @@ async def add( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - args = ["add"] - if not files: - args.append("-A" if all else ".") - else: - args.append("--") - args.extend(files) + args = build_add_args(files, all) return await self._run_git( args, path, envs, user, cwd, timeout, request_timeout ) @@ -677,16 +653,71 @@ async def commit( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - args = ["commit", "-m", message] - if allow_empty: - args.append("--allow-empty") - author_args: List[str] = [] - if author_name: - author_args.extend(["-c", f"user.name={author_name}"]) - if author_email: - author_args.extend(["-c", f"user.email={author_email}"]) - if author_args: - args = author_args + args + args = build_commit_args(message, author_name, author_email, allow_empty) + return await self._run_git( + args, path, envs, user, cwd, timeout, request_timeout + ) + + async def reset( + self, + path: str, + mode: Optional[str] = None, + target: Optional[str] = None, + paths: Optional[List[str]] = None, + envs: Optional[Dict[str, str]] = None, + user: Optional[str] = None, + cwd: Optional[str] = None, + timeout: Optional[float] = None, + request_timeout: Optional[float] = None, + ): + """ + Reset the current HEAD to a specified state. + + :param path: Repository path + :param mode: Reset mode (soft, mixed, hard, merge, keep) + :param target: Commit, branch, or ref to reset to (defaults to HEAD) + :param paths: Paths to reset + :param envs: Environment variables used for the command + :param user: User to run the command as + :param cwd: Working directory to run the command + :param timeout: Timeout for the command connection in **seconds** + :param request_timeout: Timeout for the request in **seconds** + :return: Command result from the command runner + """ + args = build_reset_args(mode, target, paths) + return await self._run_git( + args, path, envs, user, cwd, timeout, request_timeout + ) + + async def restore( + self, + path: str, + paths: List[str], + staged: Optional[bool] = None, + worktree: Optional[bool] = None, + source: Optional[str] = None, + envs: Optional[Dict[str, str]] = None, + user: Optional[str] = None, + cwd: Optional[str] = None, + timeout: Optional[float] = None, + request_timeout: Optional[float] = None, + ): + """ + Restore working tree files or unstage changes. + + :param path: Repository path + :param paths: Paths to restore (use ["."] for all) + :param staged: When True, restore the index (unstage) + :param worktree: When True, restore working tree files + :param source: Restore from the given source (commit, branch, or ref) + :param envs: Environment variables used for the command + :param user: User to run the command as + :param cwd: Working directory to run the command + :param timeout: Timeout for the command connection in **seconds** + :param request_timeout: Timeout for the request in **seconds** + :return: Command result from the command runner + """ + args = build_restore_args(paths, staged, worktree, source) return await self._run_git( args, path, envs, user, cwd, timeout, request_timeout ) @@ -822,15 +853,6 @@ async def pull( if not has_upstream: raise GitUpstreamException(build_upstream_error_message("pull")) - def build_args(remote_name: Optional[str] = None) -> List[str]: - args = ["pull"] - target_remote = remote_name or remote - if target_remote: - args.append(target_remote) - if branch: - args.append(branch) - return args - if username and password: remote_name = await self._resolve_remote_name( path, remote, envs, user, cwd, timeout, request_timeout @@ -846,7 +868,7 @@ def build_args(remote_name: Optional[str] = None) -> List[str]: timeout, request_timeout, operation=lambda: self._run_git( - build_args(remote_name), + build_pull_args(remote, branch, remote_name), path, envs, user, @@ -858,7 +880,13 @@ def build_args(remote_name: Optional[str] = None) -> List[str]: try: return await self._run_git( - build_args(), path, envs, user, cwd, timeout, request_timeout + build_pull_args(remote, branch), + path, + envs, + user, + cwd, + timeout, + request_timeout, ) except CommandExitException as err: if is_auth_failure(err): @@ -993,19 +1021,6 @@ async def dangerously_authenticate( "Both username and password are required to authenticate git." ) - target_host = host.strip() or "github.com" - target_protocol = protocol.strip() or "https" - credential_input = "\n".join( - [ - f"protocol={target_protocol}", - f"host={target_host}", - f"username={username}", - f"password={password}", - "", - "", - ] - ) - await self.set_config( "credential.helper", "store", @@ -1016,9 +1031,11 @@ async def dangerously_authenticate( timeout=timeout, request_timeout=request_timeout, ) - approve_cmd = ( - f"printf %s {shell_escape(credential_input)} | " - f"{build_git_command(['credential', 'approve'])}" + approve_cmd = build_credential_approve_command( + username=username, + password=password, + host=host, + protocol=protocol, ) return await self._run_shell( approve_cmd, diff --git a/packages/python-sdk/e2b/sandbox_sync/git.py b/packages/python-sdk/e2b/sandbox_sync/git.py index 2e0653963f..96e7628e5a 100644 --- a/packages/python-sdk/e2b/sandbox_sync/git.py +++ b/packages/python-sdk/e2b/sandbox_sync/git.py @@ -1,20 +1,36 @@ from typing import Dict, List, Optional -from e2b.sandbox.git_utils import ( +from e2b.sandbox._git import ( GitBranches, GitStatus, + build_add_args, build_auth_error_message, + build_branches_args, + build_checkout_branch_args, + build_clone_plan, + build_commit_args, + build_credential_approve_command, + build_create_branch_args, + build_delete_branch_args, build_git_command, + build_has_upstream_args, + build_pull_args, build_push_args, + build_remote_add_args, + build_remote_add_shell_command, + build_remote_get_command, + build_remote_get_url_args, + build_remote_set_url_args, + build_reset_args, + build_restore_args, + build_status_args, build_upstream_error_message, is_auth_failure, is_missing_upstream, parse_git_branches, parse_git_status, + parse_remote_url, resolve_config_scope, - shell_escape, - strip_credentials, - derive_repo_dir_from_url, with_credentials, ) from e2b.exceptions import ( @@ -116,7 +132,7 @@ def _has_upstream( ) -> bool: try: result = self._run_git( - ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"], + build_has_upstream_args(), path, envs, user, @@ -139,7 +155,7 @@ def _get_remote_url( request_timeout: Optional[float] = None, ) -> str: result = self._run_git( - ["remote", "get-url", remote], + build_remote_get_url_args(remote), path, envs, user, @@ -147,12 +163,7 @@ def _get_remote_url( timeout, request_timeout, ) - url = result.stdout.strip() - if not url: - raise InvalidArgumentException( - f'Remote "{remote}" URL not found in repository.' - ) - return url + return parse_remote_url(result.stdout, remote) def _resolve_remote_name( self, @@ -279,36 +290,22 @@ def clone( ) def attempt_clone(auth_username: Optional[str], auth_password: Optional[str]): - clone_url = ( - with_credentials(url, auth_username, auth_password) - if auth_username and auth_password - else url - ) - sanitized_url = strip_credentials(clone_url) - should_strip = ( - not dangerously_store_credentials and sanitized_url != clone_url + plan = build_clone_plan( + url=url, + path=path, + branch=branch, + depth=depth, + auth_username=auth_username, + auth_password=auth_password, + dangerously_store_credentials=dangerously_store_credentials, ) - repo_path = ( - path if not should_strip else path or derive_repo_dir_from_url(url) - ) - if should_strip and not repo_path: - raise InvalidArgumentException( - "A destination path is required when using credentials without storing them." - ) - args = ["clone", clone_url] - if branch: - args.extend(["--branch", branch, "--single-branch"]) - if depth: - args.extend(["--depth", str(depth)]) - if path: - args.append(path) result = self._run_git( - args, None, envs, user, cwd, timeout, request_timeout + plan.args, None, envs, user, cwd, timeout, request_timeout ) - if should_strip and repo_path: + if plan.should_strip and plan.repo_path and plan.sanitized_url: self._run_git( - ["remote", "set-url", "origin", sanitized_url], - repo_path, + build_remote_set_url_args("origin", plan.sanitized_url), + plan.repo_path, envs, user, cwd, @@ -386,25 +383,12 @@ def remote_add( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - if not name or not url: - raise InvalidArgumentException( - "Both remote name and URL are required to add a git remote." - ) - - args = ["remote", "add"] - if fetch: - args.append("-f") - args.extend([name, url]) + args = build_remote_add_args(name, url, fetch) if not overwrite: return self._run_git(args, path, envs, user, cwd, timeout, request_timeout) - add_cmd = build_git_command(args, path) - set_url_cmd = build_git_command(["remote", "set-url", name, url], path) - cmd = f"{add_cmd} || {set_url_cmd}" - if fetch: - fetch_cmd = build_git_command(["fetch", name], path) - cmd = f"({cmd}) && {fetch_cmd}" + cmd = build_remote_add_shell_command(args, path, name, url, fetch) return self._run_shell( cmd, envs, @@ -438,10 +422,7 @@ def remote_get( :param request_timeout: Timeout for the request in **seconds** :return: Remote URL if present, otherwise `None` """ - if not name: - raise InvalidArgumentException("Remote name is required.") - - cmd = f"{build_git_command(['remote', 'get-url', name], path)} || true" + cmd = build_remote_get_command(path, name) result = self._run_shell( cmd, envs, @@ -473,7 +454,7 @@ def status( :return: Parsed git status """ result = self._run_git( - ["status", "--porcelain=1", "-b"], + build_status_args(), path, envs, user, @@ -504,7 +485,7 @@ def branches( :return: Parsed branch list """ result = self._run_git( - ["branch", "--format=%(refname:short)\t%(HEAD)"], + build_branches_args(), path, envs, user, @@ -537,7 +518,7 @@ def create_branch( :return: Command result from the command runner """ return self._run_git( - ["checkout", "-b", branch], + build_create_branch_args(branch), path, envs, user, @@ -569,7 +550,7 @@ def checkout_branch( :return: Command result from the command runner """ return self._run_git( - ["checkout", branch], + build_checkout_branch_args(branch), path, envs, user, @@ -602,7 +583,7 @@ def delete_branch( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - args = ["branch", "-D" if force else "-d", branch] + args = build_delete_branch_args(branch, force) return self._run_git(args, path, envs, user, cwd, timeout, request_timeout) def add( @@ -629,12 +610,7 @@ def add( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - args = ["add"] - if not files: - args.append("-A" if all else ".") - else: - args.append("--") - args.extend(files) + args = build_add_args(files, all) return self._run_git(args, path, envs, user, cwd, timeout, request_timeout) def commit( @@ -665,16 +641,7 @@ def commit( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - args = ["commit", "-m", message] - if allow_empty: - args.append("--allow-empty") - author_args: List[str] = [] - if author_name: - author_args.extend(["-c", f"user.name={author_name}"]) - if author_email: - author_args.extend(["-c", f"user.email={author_email}"]) - if author_args: - args = author_args + args + args = build_commit_args(message, author_name, author_email, allow_empty) return self._run_git(args, path, envs, user, cwd, timeout, request_timeout) def reset( @@ -703,20 +670,7 @@ def reset( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - allowed_modes = {"soft", "mixed", "hard", "merge", "keep"} - if mode and mode not in allowed_modes: - raise InvalidArgumentException( - f"Reset mode must be one of {', '.join(sorted(allowed_modes))}." - ) - - args = ["reset"] - if mode: - args.append(f"--{mode}") - if target: - args.append(target) - if paths: - args.append("--") - args.extend(paths) + args = build_reset_args(mode, target, paths) return self._run_git(args, path, envs, user, cwd, timeout, request_timeout) def restore( @@ -747,32 +701,7 @@ def restore( :param request_timeout: Timeout for the request in **seconds** :return: Command result from the command runner """ - if not paths: - raise InvalidArgumentException("At least one path is required.") - - resolved_staged = staged - resolved_worktree = worktree - if staged is None and worktree is None: - resolved_worktree = True - elif staged is True and worktree is None: - resolved_worktree = False - elif staged is None and worktree is not None: - resolved_staged = False - - if resolved_staged is False and resolved_worktree is False: - raise InvalidArgumentException( - "At least one of staged or worktree must be true." - ) - - args = ["restore"] - if resolved_worktree: - args.append("--worktree") - if resolved_staged: - args.append("--staged") - if source: - args.extend(["--source", source]) - args.append("--") - args.extend(paths) + args = build_restore_args(paths, staged, worktree, source) return self._run_git(args, path, envs, user, cwd, timeout, request_timeout) def push( @@ -903,15 +832,6 @@ def pull( if not self._has_upstream(path, envs, user, cwd, timeout, request_timeout): raise GitUpstreamException(build_upstream_error_message("pull")) - def build_args(remote_name: Optional[str] = None) -> List[str]: - args = ["pull"] - target_remote = remote_name or remote - if target_remote: - args.append(target_remote) - if branch: - args.append(branch) - return args - if username and password: remote_name = self._resolve_remote_name( path, remote, envs, user, cwd, timeout, request_timeout @@ -927,7 +847,7 @@ def build_args(remote_name: Optional[str] = None) -> List[str]: timeout, request_timeout, operation=lambda: self._run_git( - build_args(remote_name), + build_pull_args(remote, branch, remote_name), path, envs, user, @@ -939,7 +859,13 @@ def build_args(remote_name: Optional[str] = None) -> List[str]: try: return self._run_git( - build_args(), path, envs, user, cwd, timeout, request_timeout + build_pull_args(remote, branch), + path, + envs, + user, + cwd, + timeout, + request_timeout, ) except CommandExitException as err: if is_auth_failure(err): @@ -1072,19 +998,6 @@ def dangerously_authenticate( "Both username and password are required to authenticate git." ) - target_host = host.strip() or "github.com" - target_protocol = protocol.strip() or "https" - credential_input = "\n".join( - [ - f"protocol={target_protocol}", - f"host={target_host}", - f"username={username}", - f"password={password}", - "", - "", - ] - ) - self.set_config( "credential.helper", "store", @@ -1095,9 +1008,11 @@ def dangerously_authenticate( timeout=timeout, request_timeout=request_timeout, ) - approve_cmd = ( - f"printf %s {shell_escape(credential_input)} | " - f"{build_git_command(['credential', 'approve'])}" + approve_cmd = build_credential_approve_command( + username=username, + password=password, + host=host, + protocol=protocol, ) return self._run_shell( approve_cmd, diff --git a/packages/python-sdk/tests/shared/git/test_parity.py b/packages/python-sdk/tests/shared/git/test_parity.py new file mode 100644 index 0000000000..c45b8f9293 --- /dev/null +++ b/packages/python-sdk/tests/shared/git/test_parity.py @@ -0,0 +1,33 @@ +import inspect + +from e2b.sandbox_async.git import Git as AsyncGit +from e2b.sandbox_sync.git import Git as SyncGit + + +def _public_methods(cls): + return { + name: getattr(cls, name) + for name in sorted(dir(cls)) + if not name.startswith("_") and callable(getattr(cls, name)) + } + + +def test_identical_method_signatures(): + sync = _public_methods(SyncGit) + async_ = _public_methods(AsyncGit) + + assert set(sync) == set(async_), ( + f"missing from async: {set(sync) - set(async_)}, " + f"missing from sync: {set(async_) - set(sync)}" + ) + + for name in sync: + assert inspect.signature(sync[name]) == inspect.signature(async_[name]), ( + f"{name}: sync{inspect.signature(sync[name])} " + f"!= async{inspect.signature(async_[name])}" + ) + + +def test_async_methods_are_coroutines(): + for name, method in _public_methods(AsyncGit).items(): + assert inspect.iscoroutinefunction(method), f"AsyncGit.{name} is not async"