diff --git a/src/pkgmgr/actions/branch/close_branch.py b/src/pkgmgr/actions/branch/close_branch.py index 991f1ca..c15e3fa 100644 --- a/src/pkgmgr/actions/branch/close_branch.py +++ b/src/pkgmgr/actions/branch/close_branch.py @@ -1,7 +1,21 @@ from __future__ import annotations + from typing import Optional -from pkgmgr.core.git import run_git, GitError, get_current_branch -from .utils import _resolve_base_branch + +from pkgmgr.core.git.errors import GitError +from pkgmgr.core.git.queries import get_current_branch +from pkgmgr.core.git.commands import ( + GitDeleteRemoteBranchError, + checkout, + delete_local_branch, + delete_remote_branch, + fetch, + merge_no_ff, + pull, + push, +) + +from pkgmgr.core.git.queries import resolve_base_branch def close_branch( @@ -14,7 +28,6 @@ def close_branch( """ Merge a feature branch into the base branch and delete it afterwards. """ - # Determine branch name if not name: try: @@ -25,7 +38,7 @@ def close_branch( if not name: raise RuntimeError("Branch name must not be empty.") - target_base = _resolve_base_branch(base_branch, fallback_base, cwd=cwd) + target_base = resolve_base_branch(base_branch, fallback_base, cwd=cwd) if name == target_base: raise RuntimeError( @@ -42,58 +55,20 @@ def close_branch( print("Aborted closing branch.") return - # Fetch - try: - run_git(["fetch", "origin"], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to fetch from origin before closing branch {name!r}: {exc}" - ) from exc + # Execute workflow (commands raise specific GitError subclasses) + fetch("origin", cwd=cwd) + checkout(target_base, cwd=cwd) + pull("origin", target_base, cwd=cwd) + merge_no_ff(name, cwd=cwd) + push("origin", target_base, cwd=cwd) - # Checkout base - try: - run_git(["checkout", target_base], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to checkout base branch {target_base!r}: {exc}" - ) from exc + # Delete local branch (safe delete by default) + delete_local_branch(name, cwd=cwd, force=False) - # Pull latest + # Delete remote branch (special-case error message) try: - run_git(["pull", "origin", target_base], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to pull latest changes for base branch {target_base!r}: {exc}" - ) from exc - - # Merge - try: - run_git(["merge", "--no-ff", name], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to merge branch {name!r} into {target_base!r}: {exc}" - ) from exc - - # Push result - try: - run_git(["push", "origin", target_base], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to push base branch {target_base!r} after merge: {exc}" - ) from exc - - # Delete local - try: - run_git(["branch", "-d", name], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to delete local branch {name!r}: {exc}" - ) from exc - - # Delete remote - try: - run_git(["push", "origin", "--delete", name], cwd=cwd) - except GitError as exc: + delete_remote_branch("origin", name, cwd=cwd) + except GitDeleteRemoteBranchError as exc: raise RuntimeError( f"Branch {name!r} deleted locally, but remote deletion failed: {exc}" ) from exc diff --git a/src/pkgmgr/actions/branch/drop_branch.py b/src/pkgmgr/actions/branch/drop_branch.py index 420f300..f6a678c 100644 --- a/src/pkgmgr/actions/branch/drop_branch.py +++ b/src/pkgmgr/actions/branch/drop_branch.py @@ -1,7 +1,16 @@ from __future__ import annotations + from typing import Optional -from pkgmgr.core.git import run_git, GitError, get_current_branch -from .utils import _resolve_base_branch + +from pkgmgr.core.git.errors import GitError +from pkgmgr.core.git.queries import get_current_branch +from pkgmgr.core.git.commands import ( + GitDeleteRemoteBranchError, + delete_local_branch, + delete_remote_branch, +) + +from pkgmgr.core.git.queries import resolve_base_branch def drop_branch( @@ -14,7 +23,6 @@ def drop_branch( """ Delete a branch locally and remotely without merging. """ - if not name: try: name = get_current_branch(cwd=cwd) @@ -24,7 +32,7 @@ def drop_branch( if not name: raise RuntimeError("Branch name must not be empty.") - target_base = _resolve_base_branch(base_branch, fallback_base, cwd=cwd) + target_base = resolve_base_branch(base_branch, fallback_base, cwd=cwd) if name == target_base: raise RuntimeError( @@ -40,16 +48,12 @@ def drop_branch( print("Aborted dropping branch.") return - # Local delete + delete_local_branch(name, cwd=cwd, force=False) + + # Remote delete (special-case message) try: - run_git(["branch", "-d", name], cwd=cwd) - except GitError as exc: - raise RuntimeError(f"Failed to delete local branch {name!r}: {exc}") from exc - - # Remote delete - try: - run_git(["push", "origin", "--delete", name], cwd=cwd) - except GitError as exc: + delete_remote_branch("origin", name, cwd=cwd) + except GitDeleteRemoteBranchError as exc: raise RuntimeError( f"Branch {name!r} was deleted locally, but remote deletion failed: {exc}" ) from exc diff --git a/src/pkgmgr/actions/branch/open_branch.py b/src/pkgmgr/actions/branch/open_branch.py index 90c511f..7531cc2 100644 --- a/src/pkgmgr/actions/branch/open_branch.py +++ b/src/pkgmgr/actions/branch/open_branch.py @@ -1,7 +1,15 @@ from __future__ import annotations + from typing import Optional -from pkgmgr.core.git import run_git, GitError -from .utils import _resolve_base_branch + +from pkgmgr.core.git.commands import ( + checkout, + create_branch, + fetch, + pull, + push_upstream, +) +from pkgmgr.core.git.queries import resolve_base_branch def open_branch( @@ -13,7 +21,6 @@ def open_branch( """ Create and push a new feature branch on top of a base branch. """ - # Request name interactively if not provided if not name: name = input("Enter new branch name: ").strip() @@ -21,44 +28,13 @@ def open_branch( if not name: raise RuntimeError("Branch name must not be empty.") - resolved_base = _resolve_base_branch(base_branch, fallback_base, cwd=cwd) + resolved_base = resolve_base_branch(base_branch, fallback_base, cwd=cwd) - # 1) Fetch from origin - try: - run_git(["fetch", "origin"], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to fetch from origin before creating branch {name!r}: {exc}" - ) from exc + # Workflow (commands raise specific GitError subclasses) + fetch("origin", cwd=cwd) + checkout(resolved_base, cwd=cwd) + pull("origin", resolved_base, cwd=cwd) - # 2) Checkout base branch - try: - run_git(["checkout", resolved_base], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to checkout base branch {resolved_base!r}: {exc}" - ) from exc - - # 3) Pull latest changes - try: - run_git(["pull", "origin", resolved_base], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to pull latest changes for base branch {resolved_base!r}: {exc}" - ) from exc - - # 4) Create new branch - try: - run_git(["checkout", "-b", name], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to create new branch {name!r} from base {resolved_base!r}: {exc}" - ) from exc - - # 5) Push new branch - try: - run_git(["push", "-u", "origin", name], cwd=cwd) - except GitError as exc: - raise RuntimeError( - f"Failed to push new branch {name!r} to origin: {exc}" - ) from exc + # Create new branch from resolved base and push it with upstream tracking + create_branch(name, resolved_base, cwd=cwd) + push_upstream("origin", name, cwd=cwd) diff --git a/src/pkgmgr/actions/branch/utils.py b/src/pkgmgr/actions/branch/utils.py deleted file mode 100644 index c37c1e1..0000000 --- a/src/pkgmgr/actions/branch/utils.py +++ /dev/null @@ -1,27 +0,0 @@ -from __future__ import annotations -from pkgmgr.core.git import run_git, GitError - - -def _resolve_base_branch( - preferred: str, - fallback: str, - cwd: str, -) -> str: - """ - Resolve the base branch to use. - - Try `preferred` first (default: main), - fall back to `fallback` (default: master). - - Raise RuntimeError if neither exists. - """ - for candidate in (preferred, fallback): - try: - run_git(["rev-parse", "--verify", candidate], cwd=cwd) - return candidate - except GitError: - continue - - raise RuntimeError( - f"Neither {preferred!r} nor {fallback!r} exist in this repository." - ) diff --git a/src/pkgmgr/actions/changelog/__init__.py b/src/pkgmgr/actions/changelog/__init__.py index 18b2357..d3557cc 100644 --- a/src/pkgmgr/actions/changelog/__init__.py +++ b/src/pkgmgr/actions/changelog/__init__.py @@ -3,17 +3,16 @@ """ Helpers to generate changelog information from Git history. - -This module provides a small abstraction around `git log` so that -CLI commands can request a changelog between two refs (tags, branches, -commits) without dealing with raw subprocess calls. """ from __future__ import annotations from typing import Optional -from pkgmgr.core.git import run_git, GitError +from pkgmgr.core.git.queries import ( + get_changelog, + GitChangelogQueryError, +) def generate_changelog( @@ -25,48 +24,20 @@ def generate_changelog( """ Generate a plain-text changelog between two Git refs. - Parameters - ---------- - cwd: - Repository directory in which to run Git commands. - from_ref: - Optional starting reference (exclusive). If provided together - with `to_ref`, the range `from_ref..to_ref` is used. - If only `from_ref` is given, the range `from_ref..HEAD` is used. - to_ref: - Optional end reference (inclusive). If omitted, `HEAD` is used. - include_merges: - If False (default), merge commits are filtered out. - - Returns - ------- - str - The output of `git log` formatted as a simple text changelog. - If no commits are found or Git fails, an explanatory message - is returned instead of raising. + Returns a human-readable message instead of raising. """ - # Determine the revision range if to_ref is None: to_ref = "HEAD" - if from_ref: - rev_range = f"{from_ref}..{to_ref}" - else: - rev_range = to_ref - - # Use a custom pretty format that includes tags/refs (%d) - cmd = [ - "log", - "--pretty=format:%h %d %s", - ] - if not include_merges: - cmd.append("--no-merges") - cmd.append(rev_range) - + rev_range = f"{from_ref}..{to_ref}" if from_ref else to_ref try: - output = run_git(cmd, cwd=cwd) - except GitError as exc: - # Do not raise to the CLI, return a human-readable error instead. + output = get_changelog( + cwd=cwd, + from_ref=from_ref, + to_ref=to_ref, + include_merges=include_merges, + ) + except GitChangelogQueryError as exc: return ( f"[ERROR] Failed to generate changelog in {cwd!r} " f"for range {rev_range!r}:\n{exc}" diff --git a/src/pkgmgr/actions/mirror/git_remote.py b/src/pkgmgr/actions/mirror/git_remote.py index 15821f2..edcbd34 100644 --- a/src/pkgmgr/actions/mirror/git_remote.py +++ b/src/pkgmgr/actions/mirror/git_remote.py @@ -1,10 +1,21 @@ from __future__ import annotations import os -from typing import List, Optional, Set +from typing import Optional, Set -from pkgmgr.core.command.run import run_command -from pkgmgr.core.git import GitError, run_git +from pkgmgr.core.git.errors import GitError +from pkgmgr.core.git.commands import ( + GitAddRemoteError, + GitAddRemotePushUrlError, + GitSetRemoteUrlError, + add_remote, + add_remote_push_url, + set_remote_url, +) +from pkgmgr.core.git.queries import ( + get_remote_push_urls, + list_remotes, +) from .types import MirrorMap, RepoMirrorContext, Repository @@ -48,29 +59,20 @@ def determine_primary_remote_url( return build_default_ssh_url(repo) -def _safe_git_output(args: List[str], cwd: str) -> Optional[str]: - try: - return run_git(args, cwd=cwd) - except GitError: - return None - - def has_origin_remote(repo_dir: str) -> bool: - out = _safe_git_output(["remote"], cwd=repo_dir) - return bool(out and "origin" in out.split()) + try: + return "origin" in list_remotes(cwd=repo_dir) + except GitError: + return False def _set_origin_fetch_and_push(repo_dir: str, url: str, preview: bool) -> None: - fetch = f"git remote set-url origin {url}" - push = f"git remote set-url --push origin {url}" - - if preview: - print(f"[PREVIEW] Would run in {repo_dir!r}: {fetch}") - print(f"[PREVIEW] Would run in {repo_dir!r}: {push}") - return - - run_command(fetch, cwd=repo_dir, preview=False) - run_command(push, cwd=repo_dir, preview=False) + """ + Ensure origin has fetch URL and push URL set to the primary URL. + Preview is handled by the underlying git runner. + """ + set_remote_url("origin", url, cwd=repo_dir, push=False, preview=preview) + set_remote_url("origin", url, cwd=repo_dir, push=True, preview=preview) def _ensure_additional_push_urls( @@ -79,22 +81,21 @@ def _ensure_additional_push_urls( primary: str, preview: bool, ) -> None: + """ + Ensure all mirror URLs (except primary) are configured as additional push URLs for origin. + Preview is handled by the underlying git runner. + """ desired: Set[str] = {u for u in mirrors.values() if u and u != primary} if not desired: return - out = _safe_git_output( - ["remote", "get-url", "--push", "--all", "origin"], - cwd=repo_dir, - ) - existing = set(out.splitlines()) if out else set() + try: + existing = get_remote_push_urls("origin", cwd=repo_dir) + except GitError: + existing = set() for url in sorted(desired - existing): - cmd = f"git remote set-url --add --push origin {url}" - if preview: - print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}") - else: - run_command(cmd, cwd=repo_dir, preview=False) + add_remote_push_url("origin", url, cwd=repo_dir, preview=preview) def ensure_origin_remote( @@ -113,21 +114,23 @@ def ensure_origin_remote( print("[WARN] No primary mirror URL could be determined.") return + # 1) Ensure origin exists if not has_origin_remote(repo_dir): - cmd = f"git remote add origin {primary}" - if preview: - print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}") - else: - run_command(cmd, cwd=repo_dir, preview=False) + try: + add_remote("origin", primary, cwd=repo_dir, preview=preview) + except GitAddRemoteError as exc: + print(f"[WARN] Failed to add origin remote: {exc}") + return # without origin we cannot reliably proceed - _set_origin_fetch_and_push(repo_dir, primary, preview) - - _ensure_additional_push_urls(repo_dir, ctx.resolved_mirrors, primary, preview) - - -def is_remote_reachable(url: str, cwd: Optional[str] = None) -> bool: + # 2) Ensure origin fetch+push URLs are correct (ALWAYS, even if origin already existed) try: - run_git(["ls-remote", "--exit-code", url], cwd=cwd or os.getcwd()) - return True - except GitError: - return False + _set_origin_fetch_and_push(repo_dir, primary, preview) + except GitSetRemoteUrlError as exc: + # Do not abort: still try to add additional push URLs + print(f"[WARN] Failed to set origin URLs: {exc}") + + # 3) Ensure additional push URLs for mirrors + try: + _ensure_additional_push_urls(repo_dir, ctx.resolved_mirrors, primary, preview) + except GitAddRemotePushUrlError as exc: + print(f"[WARN] Failed to add additional push URLs: {exc}") diff --git a/src/pkgmgr/actions/mirror/remote_check.py b/src/pkgmgr/actions/mirror/remote_check.py deleted file mode 100644 index beb3127..0000000 --- a/src/pkgmgr/actions/mirror/remote_check.py +++ /dev/null @@ -1,21 +0,0 @@ -# src/pkgmgr/actions/mirror/remote_check.py -from __future__ import annotations - -from typing import Tuple - -from pkgmgr.core.git import GitError, run_git - - -def probe_mirror(url: str, repo_dir: str) -> Tuple[bool, str]: - """ - Probe a remote mirror URL using `git ls-remote`. - - Returns: - (True, "") on success, - (False, error_message) on failure. - """ - try: - run_git(["ls-remote", url], cwd=repo_dir) - return True, "" - except GitError as exc: - return False, str(exc) diff --git a/src/pkgmgr/actions/mirror/setup_cmd.py b/src/pkgmgr/actions/mirror/setup_cmd.py index 37a63c9..77d58b2 100644 --- a/src/pkgmgr/actions/mirror/setup_cmd.py +++ b/src/pkgmgr/actions/mirror/setup_cmd.py @@ -4,7 +4,7 @@ from typing import List from .context import build_context from .git_remote import ensure_origin_remote, determine_primary_remote_url -from .remote_check import probe_mirror +from pkgmgr.core.git.queries import probe_remote_reachable from .remote_provision import ensure_remote_repository from .types import Repository @@ -52,19 +52,14 @@ def _setup_remote_mirrors_for_repo( primary = determine_primary_remote_url(repo, ctx) if not primary: return - - ok, msg = probe_mirror(primary, ctx.repo_dir) + ok = probe_remote_reachable(primary, cwd=ctx.repo_dir) print("[OK]" if ok else "[WARN]", primary) - if msg: - print(msg) print() return for name, url in ctx.resolved_mirrors.items(): - ok, msg = probe_mirror(url, ctx.repo_dir) + ok = probe_remote_reachable(url, cwd=ctx.repo_dir) print(f"[OK] {name}: {url}" if ok else f"[WARN] {name}: {url}") - if msg: - print(msg) print() diff --git a/src/pkgmgr/actions/publish/git_tags.py b/src/pkgmgr/actions/publish/git_tags.py index 6f7fa09..de33e75 100644 --- a/src/pkgmgr/actions/publish/git_tags.py +++ b/src/pkgmgr/actions/publish/git_tags.py @@ -1,17 +1,10 @@ from __future__ import annotations -from pkgmgr.core.git import run_git +from pkgmgr.core.git.queries import get_tags_at_ref from pkgmgr.core.version.semver import SemVer, is_semver_tag def head_semver_tags(cwd: str = ".") -> list[str]: - out = run_git(["tag", "--points-at", "HEAD"], cwd=cwd) - if not out: - return [] - - tags = [t.strip() for t in out.splitlines() if t.strip()] + tags = get_tags_at_ref("HEAD", cwd=cwd) tags = [t for t in tags if is_semver_tag(t) and t.startswith("v")] - if not tags: - return [] - return sorted(tags, key=SemVer.parse) diff --git a/src/pkgmgr/core/git/__init__.py b/src/pkgmgr/core/git/__init__.py index 2915cd3..5c494ae 100644 --- a/src/pkgmgr/core/git/__init__.py +++ b/src/pkgmgr/core/git/__init__.py @@ -1,5 +1,4 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- +from __future__ import annotations """ Lightweight helper functions around Git commands. @@ -9,84 +8,10 @@ logic (release, version, changelog) does not have to deal with the details of subprocess handling. """ -from __future__ import annotations +from .errors import GitError +from .run import run -import subprocess -from typing import List, Optional - - -class GitError(RuntimeError): - """Raised when a Git command fails in an unexpected way.""" - - -def run_git(args: List[str], cwd: str = ".") -> str: - """ - Run a Git command and return its stdout as a stripped string. - - Raises GitError if the command fails. - """ - cmd = ["git"] + args - try: - result = subprocess.run( - cmd, - cwd=cwd, - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) - except subprocess.CalledProcessError as exc: - raise GitError( - f"Git command failed in {cwd!r}: {' '.join(cmd)}\n" - f"Exit code: {exc.returncode}\n" - f"STDOUT:\n{exc.stdout}\n" - f"STDERR:\n{exc.stderr}" - ) from exc - - return result.stdout.strip() - - -def get_tags(cwd: str = ".") -> List[str]: - """ - Return a list of all tags in the repository in `cwd`. - - If there are no tags, an empty list is returned. - """ - try: - output = run_git(["tag"], cwd=cwd) - except GitError as exc: - # If the repo has no tags or is not a git repo, surface a clear error. - # You can decide later if you want to treat this differently. - if "not a git repository" in str(exc): - raise - # No tags: stdout may just be empty; treat this as empty list. - return [] - - if not output: - return [] - - return [line.strip() for line in output.splitlines() if line.strip()] - - -def get_head_commit(cwd: str = ".") -> Optional[str]: - """ - Return the current HEAD commit hash, or None if it cannot be determined. - """ - try: - output = run_git(["rev-parse", "HEAD"], cwd=cwd) - except GitError: - return None - return output or None - - -def get_current_branch(cwd: str = ".") -> Optional[str]: - """ - Return the current branch name, or None if it cannot be determined. - - Note: In detached HEAD state this will return 'HEAD'. - """ - try: - output = run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=cwd) - except GitError: - return None - return output or None +__all__ = [ + "GitError", + "run" +] diff --git a/src/pkgmgr/core/git/commands/__init__.py b/src/pkgmgr/core/git/commands/__init__.py new file mode 100644 index 0000000..0787e48 --- /dev/null +++ b/src/pkgmgr/core/git/commands/__init__.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from .checkout import GitCheckoutError, checkout +from .delete_local_branch import GitDeleteLocalBranchError, delete_local_branch +from .delete_remote_branch import GitDeleteRemoteBranchError, delete_remote_branch +from .fetch import GitFetchError, fetch +from .merge_no_ff import GitMergeError, merge_no_ff +from .pull import GitPullError, pull +from .push import GitPushError, push +from .create_branch import GitCreateBranchError, create_branch +from .push_upstream import GitPushUpstreamError, push_upstream + +from .add_remote import GitAddRemoteError, add_remote +from .set_remote_url import GitSetRemoteUrlError, set_remote_url +from .add_remote_push_url import GitAddRemotePushUrlError, add_remote_push_url + +__all__ = [ + "fetch", + "checkout", + "pull", + "merge_no_ff", + "push", + "delete_local_branch", + "delete_remote_branch", + "create_branch", + "push_upstream", + "add_remote", + "set_remote_url", + "add_remote_push_url", + "GitFetchError", + "GitCheckoutError", + "GitPullError", + "GitMergeError", + "GitPushError", + "GitDeleteLocalBranchError", + "GitDeleteRemoteBranchError", + "GitCreateBranchError", + "GitPushUpstreamError", + "GitAddRemoteError", + "GitSetRemoteUrlError", + "GitAddRemotePushUrlError", +] diff --git a/src/pkgmgr/core/git/commands/add_remote.py b/src/pkgmgr/core/git/commands/add_remote.py new file mode 100644 index 0000000..72056be --- /dev/null +++ b/src/pkgmgr/core/git/commands/add_remote.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitAddRemoteError(GitCommandError): + """Raised when adding a remote fails.""" + + +def add_remote( + name: str, + url: str, + *, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Add a new remote. + + Equivalent to: + git remote add + """ + try: + run( + ["remote", "add", name, url], + cwd=cwd, + preview=preview, + ) + except GitError as exc: + raise GitAddRemoteError( + f"Failed to add remote {name!r} with URL {url!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/add_remote_push_url.py b/src/pkgmgr/core/git/commands/add_remote_push_url.py new file mode 100644 index 0000000..a401059 --- /dev/null +++ b/src/pkgmgr/core/git/commands/add_remote_push_url.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitAddRemotePushUrlError(GitCommandError): + """Raised when adding an additional push URL to a remote fails.""" + + +def add_remote_push_url( + remote: str, + url: str, + *, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Add an additional push URL to a remote. + + Equivalent to: + git remote set-url --add --push + """ + try: + run( + ["remote", "set-url", "--add", "--push", remote, url], + cwd=cwd, + preview=preview, + ) + except GitError as exc: + raise GitAddRemotePushUrlError( + f"Failed to add push url {url!r} to remote {remote!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/checkout.py b/src/pkgmgr/core/git/commands/checkout.py new file mode 100644 index 0000000..8506449 --- /dev/null +++ b/src/pkgmgr/core/git/commands/checkout.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitCheckoutError(GitCommandError): + """Raised when checking out a branch fails.""" + + +def checkout(branch: str, cwd: str = ".") -> None: + try: + run(["checkout", branch], cwd=cwd) + except GitError as exc: + raise GitCheckoutError( + f"Failed to checkout branch {branch!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/create_branch.py b/src/pkgmgr/core/git/commands/create_branch.py new file mode 100644 index 0000000..359adff --- /dev/null +++ b/src/pkgmgr/core/git/commands/create_branch.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitCreateBranchError(GitCommandError): + """Raised when creating a new branch fails.""" + + +def create_branch(branch: str, base: str, cwd: str = ".") -> None: + """ + Create a new branch from a base branch. + + Equivalent to: git checkout -b + """ + try: + run(["checkout", "-b", branch, base], cwd=cwd) + except GitError as exc: + raise GitCreateBranchError( + f"Failed to create branch {branch!r} from base {base!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/delete_local_branch.py b/src/pkgmgr/core/git/commands/delete_local_branch.py new file mode 100644 index 0000000..5c6340d --- /dev/null +++ b/src/pkgmgr/core/git/commands/delete_local_branch.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitDeleteLocalBranchError(GitCommandError): + """Raised when deleting a local branch fails.""" + + +def delete_local_branch(branch: str, cwd: str = ".", force: bool = False) -> None: + flag = "-D" if force else "-d" + try: + run(["branch", flag, branch], cwd=cwd) + except GitError as exc: + raise GitDeleteLocalBranchError( + f"Failed to delete local branch {branch!r} (flag {flag}).", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/delete_remote_branch.py b/src/pkgmgr/core/git/commands/delete_remote_branch.py new file mode 100644 index 0000000..1f9e28a --- /dev/null +++ b/src/pkgmgr/core/git/commands/delete_remote_branch.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitDeleteRemoteBranchError(GitCommandError): + """Raised when deleting a remote branch fails.""" + + +def delete_remote_branch(remote: str, branch: str, cwd: str = ".") -> None: + try: + run(["push", remote, "--delete", branch], cwd=cwd) + except GitError as exc: + raise GitDeleteRemoteBranchError( + f"Failed to delete remote branch {branch!r} on {remote!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/fetch.py b/src/pkgmgr/core/git/commands/fetch.py new file mode 100644 index 0000000..2b60d63 --- /dev/null +++ b/src/pkgmgr/core/git/commands/fetch.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitFetchError(GitCommandError): + """Raised when fetching from a remote fails.""" + + +def fetch(remote: str = "origin", cwd: str = ".") -> None: + try: + run(["fetch", remote], cwd=cwd) + except GitError as exc: + raise GitFetchError( + f"Failed to fetch from remote {remote!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/merge_no_ff.py b/src/pkgmgr/core/git/commands/merge_no_ff.py new file mode 100644 index 0000000..cc9f30d --- /dev/null +++ b/src/pkgmgr/core/git/commands/merge_no_ff.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitMergeError(GitCommandError): + """Raised when merging a branch fails.""" + + +def merge_no_ff(branch: str, cwd: str = ".") -> None: + try: + run(["merge", "--no-ff", branch], cwd=cwd) + except GitError as exc: + raise GitMergeError( + f"Failed to merge branch {branch!r} with --no-ff.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/pull.py b/src/pkgmgr/core/git/commands/pull.py new file mode 100644 index 0000000..ff5753f --- /dev/null +++ b/src/pkgmgr/core/git/commands/pull.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitPullError(GitCommandError): + """Raised when pulling from a remote branch fails.""" + + +def pull(remote: str, branch: str, cwd: str = ".") -> None: + try: + run(["pull", remote, branch], cwd=cwd) + except GitError as exc: + raise GitPullError( + f"Failed to pull {remote!r}/{branch!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/push.py b/src/pkgmgr/core/git/commands/push.py new file mode 100644 index 0000000..bbe9485 --- /dev/null +++ b/src/pkgmgr/core/git/commands/push.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitPushError(GitCommandError): + """Raised when pushing to a remote fails.""" + + +def push(remote: str, ref: str, cwd: str = ".") -> None: + try: + run(["push", remote, ref], cwd=cwd) + except GitError as exc: + raise GitPushError( + f"Failed to push ref {ref!r} to remote {remote!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/push_upstream.py b/src/pkgmgr/core/git/commands/push_upstream.py new file mode 100644 index 0000000..9c6ab82 --- /dev/null +++ b/src/pkgmgr/core/git/commands/push_upstream.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitPushUpstreamError(GitCommandError): + """Raised when pushing a branch with upstream tracking fails.""" + + +def push_upstream(remote: str, branch: str, cwd: str = ".") -> None: + """ + Push a branch and set upstream tracking. + + Equivalent to: git push -u + """ + try: + run(["push", "-u", remote, branch], cwd=cwd) + except GitError as exc: + raise GitPushUpstreamError( + f"Failed to push branch {branch!r} to {remote!r} with upstream tracking.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/set_remote_url.py b/src/pkgmgr/core/git/commands/set_remote_url.py new file mode 100644 index 0000000..61090c3 --- /dev/null +++ b/src/pkgmgr/core/git/commands/set_remote_url.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitSetRemoteUrlError(GitCommandError): + """Raised when setting a remote URL fails.""" + + +def set_remote_url( + remote: str, + url: str, + *, + cwd: str = ".", + push: bool = False, + preview: bool = False, +) -> None: + """ + Set the fetch or push URL of a remote. + + Equivalent to: + git remote set-url + or: + git remote set-url --push + """ + args = ["remote", "set-url"] + if push: + args.append("--push") + args += [remote, url] + + try: + run( + args, + cwd=cwd, + preview=preview, + ) + except GitError as exc: + mode = "push" if push else "fetch" + raise GitSetRemoteUrlError( + f"Failed to set {mode} url for remote {remote!r} to {url!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/errors.py b/src/pkgmgr/core/git/errors.py new file mode 100644 index 0000000..be192f6 --- /dev/null +++ b/src/pkgmgr/core/git/errors.py @@ -0,0 +1,16 @@ +from __future__ import annotations + + +class GitError(RuntimeError): + """Base error raised for Git related failures.""" + + +class GitCommandError(GitError): + """ + Base class for state-changing git command failures. + + Use subclasses to provide stable error types for callers. + """ + def __init__(self, message: str, *, cwd: str = ".") -> None: + super().__init__(message) + self.cwd = cwd \ No newline at end of file diff --git a/src/pkgmgr/core/git/queries/__init__.py b/src/pkgmgr/core/git/queries/__init__.py new file mode 100644 index 0000000..1a3ff9e --- /dev/null +++ b/src/pkgmgr/core/git/queries/__init__.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from .get_current_branch import get_current_branch +from .get_head_commit import get_head_commit +from .get_tags import get_tags +from .resolve_base_branch import GitBaseBranchNotFoundError, resolve_base_branch +from .list_remotes import list_remotes +from .get_remote_push_urls import get_remote_push_urls +from .probe_remote_reachable import probe_remote_reachable +from .get_changelog import get_changelog, GitChangelogQueryError +from .get_tags_at_ref import get_tags_at_ref, GitTagsAtRefQueryError + +__all__ = [ + "get_current_branch", + "get_head_commit", + "get_tags", + "resolve_base_branch", + "GitBaseBranchNotFoundError", + "list_remotes", + "get_remote_push_urls", + "probe_remote_reachable", + "get_changelog", + "GitChangelogQueryError", + "get_tags_at_ref", + "GitTagsAtRefQueryError", +] diff --git a/src/pkgmgr/core/git/queries/get_changelog.py b/src/pkgmgr/core/git/queries/get_changelog.py new file mode 100644 index 0000000..0344581 --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_changelog.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from typing import Optional + +from ..errors import GitError +from ..run import run + + +class GitChangelogQueryError(GitError): + """Raised when querying the git changelog fails.""" + + +def get_changelog( + *, + cwd: str, + from_ref: Optional[str] = None, + to_ref: Optional[str] = None, + include_merges: bool = False, +) -> str: + """ + Return a plain-text changelog between two Git refs. + + Uses: + git log --pretty=format:%h %d %s [--no-merges] + + Raises: + GitChangelogQueryError on failure. + """ + if to_ref is None: + to_ref = "HEAD" + + rev_range = f"{from_ref}..{to_ref}" if from_ref else to_ref + + cmd = ["log", "--pretty=format:%h %d %s"] + if not include_merges: + cmd.append("--no-merges") + cmd.append(rev_range) + + try: + return run(cmd, cwd=cwd) + except GitError as exc: + raise GitChangelogQueryError( + f"Failed to query changelog for range {rev_range!r}.", + ) from exc diff --git a/src/pkgmgr/core/git/queries/get_current_branch.py b/src/pkgmgr/core/git/queries/get_current_branch.py new file mode 100644 index 0000000..36db71e --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_current_branch.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from typing import Optional +from ..errors import GitError +from ..run import run + + +def get_current_branch(cwd: str = ".") -> Optional[str]: + """ + Return the current branch name, or None if it cannot be determined. + + Note: In detached HEAD state this will return 'HEAD'. + """ + try: + output = run(["rev-parse", "--abbrev-ref", "HEAD"], cwd=cwd) + except GitError: + return None + return output or None \ No newline at end of file diff --git a/src/pkgmgr/core/git/queries/get_head_commit.py b/src/pkgmgr/core/git/queries/get_head_commit.py new file mode 100644 index 0000000..c7b7e55 --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_head_commit.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from typing import Optional + +from ..errors import GitError +from ..run import run + + +def get_head_commit(cwd: str = ".") -> Optional[str]: + """ + Return the current HEAD commit hash, or None if it cannot be determined. + """ + try: + output = run(["rev-parse", "HEAD"], cwd=cwd) + except GitError: + return None + return output or None diff --git a/src/pkgmgr/core/git/queries/get_remote_push_urls.py b/src/pkgmgr/core/git/queries/get_remote_push_urls.py new file mode 100644 index 0000000..3d89e4b --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_remote_push_urls.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from typing import Set + +from ..errors import GitError +from ..run import run + + +def get_remote_push_urls(remote: str, cwd: str = ".") -> Set[str]: + """ + Return all push URLs configured for a remote. + + Equivalent to: + git remote get-url --push --all + + Raises GitError if the command fails. + """ + output = run(["remote", "get-url", "--push", "--all", remote], cwd=cwd) + if not output: + return set() + return {line.strip() for line in output.splitlines() if line.strip()} diff --git a/src/pkgmgr/core/git/queries/get_tags.py b/src/pkgmgr/core/git/queries/get_tags.py new file mode 100644 index 0000000..7ceb65a --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_tags.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from typing import List + +from ..errors import GitError +from ..run import run + + +def get_tags(cwd: str = ".") -> List[str]: + """ + Return a list of all tags in the repository in `cwd`. + + If there are no tags, an empty list is returned. + """ + try: + output = run(["tag"], cwd=cwd) + except GitError as exc: + # If the repo is not a git repo, surface a clear error. + if "not a git repository" in str(exc): + raise + # Otherwise, treat as "no tags" (e.g., empty stdout). + return [] + + if not output: + return [] + + return [line.strip() for line in output.splitlines() if line.strip()] diff --git a/src/pkgmgr/core/git/queries/get_tags_at_ref.py b/src/pkgmgr/core/git/queries/get_tags_at_ref.py new file mode 100644 index 0000000..70175f1 --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_tags_at_ref.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from typing import List + +from ..errors import GitError +from ..run import run + + +class GitTagsAtRefQueryError(GitError): + """Raised when querying tags for a ref fails.""" + + +def get_tags_at_ref(ref: str, *, cwd: str = ".") -> List[str]: + """ + Return all git tags pointing at a given ref. + + Equivalent to: + git tag --points-at + """ + try: + output = run(["tag", "--points-at", ref], cwd=cwd) + except GitError as exc: + raise GitTagsAtRefQueryError( + f"Failed to query tags at ref {ref!r}.", + ) from exc + + if not output: + return [] + + return [line.strip() for line in output.splitlines() if line.strip()] diff --git a/src/pkgmgr/core/git/queries/list_remotes.py b/src/pkgmgr/core/git/queries/list_remotes.py new file mode 100644 index 0000000..1a30444 --- /dev/null +++ b/src/pkgmgr/core/git/queries/list_remotes.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from typing import List + +from ..errors import GitError +from ..run import run + + +def list_remotes(cwd: str = ".") -> List[str]: + """ + Return a list of configured git remotes (e.g. ['origin', 'upstream']). + + Raises GitError if the command fails. + """ + output = run(["remote"], cwd=cwd) + if not output: + return [] + return [line.strip() for line in output.splitlines() if line.strip()] diff --git a/src/pkgmgr/core/git/queries/probe_remote_reachable.py b/src/pkgmgr/core/git/queries/probe_remote_reachable.py new file mode 100644 index 0000000..8133a93 --- /dev/null +++ b/src/pkgmgr/core/git/queries/probe_remote_reachable.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from ..errors import GitError +from ..run import run + + +def probe_remote_reachable(url: str, cwd: str = ".") -> bool: + """ + Check whether a remote URL is reachable. + + Equivalent to: + git ls-remote --exit-code + + Returns: + True if reachable, False otherwise. + """ + try: + run(["ls-remote", "--exit-code", url], cwd=cwd) + return True + except GitError: + return False diff --git a/src/pkgmgr/core/git/queries/resolve_base_branch.py b/src/pkgmgr/core/git/queries/resolve_base_branch.py new file mode 100644 index 0000000..b680774 --- /dev/null +++ b/src/pkgmgr/core/git/queries/resolve_base_branch.py @@ -0,0 +1,66 @@ +# src/pkgmgr/core/git/queries/resolve_base_branch.py +from __future__ import annotations + +from ..errors import GitError +from ..run import run + + +class GitBaseBranchNotFoundError(GitError): + """Raised when neither preferred nor fallback base branch exists.""" + + +def _is_branch_missing_error(exc: GitError) -> bool: + """ + Heuristic: Detect errors that indicate the branch/ref does not exist. + + We intentionally *do not* swallow other errors like: + - not a git repository + - permission issues + - corrupted repository + """ + msg = str(exc).lower() + + # Common git messages when verifying a non-existing ref/branch. + patterns = [ + "needed a single revision", + "unknown revision or path not in the working tree", + "not a valid object name", + "ambiguous argument", + "bad revision", + "fatal: invalid object name", + "fatal: ambiguous argument", + ] + + return any(p in msg for p in patterns) + + +def resolve_base_branch( + preferred: str = "main", + fallback: str = "master", + cwd: str = ".", +) -> str: + """ + Resolve the base branch to use. + + Try `preferred` first (default: main), + fall back to `fallback` (default: master). + + Raises GitBaseBranchNotFoundError if neither exists. + Raises GitError for other git failures (e.g., not a git repository). + """ + last_missing_error: GitError | None = None + + for candidate in (preferred, fallback): + try: + run(["rev-parse", "--verify", candidate], cwd=cwd) + return candidate + except GitError as exc: + if _is_branch_missing_error(exc): + last_missing_error = exc + continue + raise # anything else is a real problem -> bubble up + + # Both candidates missing -> raise specific error + raise GitBaseBranchNotFoundError( + f"Neither {preferred!r} nor {fallback!r} exist in this repository." + ) from last_missing_error diff --git a/src/pkgmgr/core/git/run.py b/src/pkgmgr/core/git/run.py new file mode 100644 index 0000000..d6d72b6 --- /dev/null +++ b/src/pkgmgr/core/git/run.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import subprocess +from typing import List + +from .errors import GitError + + +def run( + args: List[str], + *, + cwd: str = ".", + preview: bool = False, +) -> str: + """ + Run a Git command and return its stdout as a stripped string. + + If preview=True, the command is printed but NOT executed. + + Raises GitError if execution fails. + """ + cmd = ["git"] + args + cmd_str = " ".join(cmd) + + if preview: + print(f"[PREVIEW] Would run in {cwd!r}: {cmd_str}") + return "" + + try: + result = subprocess.run( + cmd, + cwd=cwd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except subprocess.CalledProcessError as exc: + raise GitError( + f"Git command failed in {cwd!r}: {cmd_str}\n" + f"Exit code: {exc.returncode}\n" + f"STDOUT:\n{exc.stdout}\n" + f"STDERR:\n{exc.stderr}" + ) from exc + + return result.stdout.strip() diff --git a/tests/unit/pkgmgr/actions/branch/__init__.py b/tests/unit/pkgmgr/actions/branch/__init__.py index ae5743f..e69de29 100644 --- a/tests/unit/pkgmgr/actions/branch/__init__.py +++ b/tests/unit/pkgmgr/actions/branch/__init__.py @@ -1,33 +0,0 @@ -import unittest -from unittest.mock import patch - -from pkgmgr.actions.branch.utils import _resolve_base_branch -from pkgmgr.core.git import GitError - - -class TestResolveBaseBranch(unittest.TestCase): - @patch("pkgmgr.actions.branch.utils.run_git") - def test_resolves_preferred(self, run_git): - run_git.return_value = None - result = _resolve_base_branch("main", "master", cwd=".") - self.assertEqual(result, "main") - run_git.assert_called_with(["rev-parse", "--verify", "main"], cwd=".") - - @patch("pkgmgr.actions.branch.utils.run_git") - def test_resolves_fallback(self, run_git): - run_git.side_effect = [ - GitError("main missing"), - None, - ] - result = _resolve_base_branch("main", "master", cwd=".") - self.assertEqual(result, "master") - - @patch("pkgmgr.actions.branch.utils.run_git") - def test_raises_when_no_branch_exists(self, run_git): - run_git.side_effect = GitError("missing") - with self.assertRaises(RuntimeError): - _resolve_base_branch("main", "master", cwd=".") - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit/pkgmgr/actions/mirror/test_remote_check.py b/tests/unit/pkgmgr/actions/mirror/test_remote_check.py deleted file mode 100644 index 040e1a8..0000000 --- a/tests/unit/pkgmgr/actions/mirror/test_remote_check.py +++ /dev/null @@ -1,52 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -from __future__ import annotations - -import unittest -from unittest.mock import patch - -from pkgmgr.actions.mirror.remote_check import probe_mirror -from pkgmgr.core.git import GitError - - -class TestRemoteCheck(unittest.TestCase): - """ - Unit tests for non-destructive remote probing (git ls-remote). - """ - - @patch("pkgmgr.actions.mirror.remote_check.run_git") - def test_probe_mirror_success_returns_true_and_empty_message(self, mock_run_git) -> None: - mock_run_git.return_value = "dummy-output" - - ok, message = probe_mirror( - "ssh://git@code.example.org:2201/alice/repo.git", - "/tmp/some-repo", - ) - - self.assertTrue(ok) - self.assertEqual(message, "") - mock_run_git.assert_called_once_with( - ["ls-remote", "ssh://git@code.example.org:2201/alice/repo.git"], - cwd="/tmp/some-repo", - ) - - @patch("pkgmgr.actions.mirror.remote_check.run_git") - def test_probe_mirror_failure_returns_false_and_error_message(self, mock_run_git) -> None: - mock_run_git.side_effect = GitError("Git command failed (simulated)") - - ok, message = probe_mirror( - "ssh://git@code.example.org:2201/alice/repo.git", - "/tmp/some-repo", - ) - - self.assertFalse(ok) - self.assertIn("Git command failed", message) - mock_run_git.assert_called_once_with( - ["ls-remote", "ssh://git@code.example.org:2201/alice/repo.git"], - cwd="/tmp/some-repo", - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit/pkgmgr/actions/branch/test_utils.py b/tests/unit/pkgmgr/core/git/__init__.py similarity index 100% rename from tests/unit/pkgmgr/actions/branch/test_utils.py rename to tests/unit/pkgmgr/core/git/__init__.py diff --git a/tests/unit/pkgmgr/core/git/queries/__init__.py b/tests/unit/pkgmgr/core/git/queries/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/pkgmgr/core/git/queries/test_remote_check.py b/tests/unit/pkgmgr/core/git/queries/test_remote_check.py new file mode 100644 index 0000000..4e3955f --- /dev/null +++ b/tests/unit/pkgmgr/core/git/queries/test_remote_check.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from pkgmgr.core.git import GitError +from pkgmgr.core.git.queries.probe_remote_reachable import probe_remote_reachable + + +class TestProbeRemoteReachable(unittest.TestCase): + """ + Unit tests for non-destructive remote probing (git ls-remote). + """ + + @patch("pkgmgr.core.git.queries.probe_remote_reachable.run") + def test_probe_remote_reachable_success_returns_true(self, mock_run) -> None: + mock_run.return_value = "dummy-output" + + ok = probe_remote_reachable( + "ssh://git@code.example.org:2201/alice/repo.git", + cwd="/tmp/some-repo", + ) + + self.assertTrue(ok) + mock_run.assert_called_once_with( + ["ls-remote", "--exit-code", "ssh://git@code.example.org:2201/alice/repo.git"], + cwd="/tmp/some-repo", + ) + + @patch("pkgmgr.core.git.queries.probe_remote_reachable.run") + def test_probe_remote_reachable_failure_returns_false(self, mock_run) -> None: + mock_run.side_effect = GitError("Git command failed (simulated)") + + ok = probe_remote_reachable( + "ssh://git@code.example.org:2201/alice/repo.git", + cwd="/tmp/some-repo", + ) + + self.assertFalse(ok) + mock_run.assert_called_once_with( + ["ls-remote", "--exit-code", "ssh://git@code.example.org:2201/alice/repo.git"], + cwd="/tmp/some-repo", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py b/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py new file mode 100644 index 0000000..d73bdc7 --- /dev/null +++ b/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from pkgmgr.core.git import GitError +from pkgmgr.core.git.queries.resolve_base_branch import ( + GitBaseBranchNotFoundError, + resolve_base_branch, +) + + +class TestResolveBaseBranch(unittest.TestCase): + @patch("pkgmgr.core.git.queries.resolve_base_branch.run") + def test_resolves_preferred(self, mock_run): + mock_run.return_value = "dummy" + result = resolve_base_branch("main", "master", cwd=".") + self.assertEqual(result, "main") + mock_run.assert_called_with(["rev-parse", "--verify", "main"], cwd=".") + + @patch("pkgmgr.core.git.queries.resolve_base_branch.run") + def test_resolves_fallback(self, mock_run): + mock_run.side_effect = [ + GitError("fatal: Needed a single revision"), # treat as "missing" + "dummy", + ] + result = resolve_base_branch("main", "master", cwd=".") + self.assertEqual(result, "master") + self.assertEqual(mock_run.call_args_list[0].kwargs["cwd"], ".") + self.assertEqual(mock_run.call_args_list[1].kwargs["cwd"], ".") + + @patch("pkgmgr.core.git.queries.resolve_base_branch.run") + def test_raises_when_no_branch_exists(self, mock_run): + mock_run.side_effect = [ + GitError("fatal: Needed a single revision"), + GitError("fatal: Needed a single revision"), + ] + with self.assertRaises(GitBaseBranchNotFoundError): + resolve_base_branch("main", "master", cwd=".") + + +if __name__ == "__main__": + unittest.main()