diff --git a/src/pkgmgr/actions/release/git_ops.py b/src/pkgmgr/actions/release/git_ops.py index b195f00..6c1441b 100644 --- a/src/pkgmgr/actions/release/git_ops.py +++ b/src/pkgmgr/actions/release/git_ops.py @@ -1,73 +1,90 @@ from __future__ import annotations -import subprocess - -from pkgmgr.core.git import GitError +from pkgmgr.core.git.commands import ( + fetch, + pull_ff_only, + push, + tag_force_annotated, +) +from pkgmgr.core.git.queries import get_upstream_ref, list_tags -def run_git_command(cmd: str) -> None: - print(f"[GIT] {cmd}") - try: - subprocess.run( - cmd, - shell=True, - check=True, - text=True, - capture_output=True, - ) - except subprocess.CalledProcessError as exc: - print(f"[ERROR] Git command failed: {cmd}") - print(f" Exit code: {exc.returncode}") - if exc.stdout: - print("\n" + exc.stdout) - if exc.stderr: - print("\n" + exc.stderr) - raise GitError(f"Git command failed: {cmd}") from exc - - -def _capture(cmd: str) -> str: - res = subprocess.run(cmd, shell=True, check=False, capture_output=True, text=True) - return (res.stdout or "").strip() - - -def ensure_clean_and_synced(preview: bool = False) -> None: +def ensure_clean_and_synced(*, preview: bool = False) -> None: """ Always run a pull BEFORE modifying anything. Uses --ff-only to avoid creating merge commits automatically. If no upstream is configured, we skip. """ - upstream = _capture("git rev-parse --abbrev-ref --symbolic-full-name @{u} 2>/dev/null") + upstream = get_upstream_ref() if not upstream: print("[INFO] No upstream configured for current branch. Skipping pull.") return - if preview: - print("[PREVIEW] Would run: git fetch origin --prune --tags --force") - print("[PREVIEW] Would run: git pull --ff-only") - return - print("[INFO] Syncing with remote before making any changes...") - run_git_command("git fetch origin --prune --tags --force") - run_git_command("git pull --ff-only") + + # Mirrors old behavior: + # git fetch origin --prune --tags --force + # git pull --ff-only + fetch(remote="origin", prune=True, tags=True, force=True, preview=preview) + pull_ff_only(preview=preview) + + +def _parse_v_tag(tag: str) -> tuple[int, ...] | None: + """ + Parse tags like 'v1.2.3' into (1, 2, 3). + Returns None if parsing is not possible. + """ + if not tag.startswith("v"): + return None + + raw = tag[1:] + if not raw: + return None + + parts = raw.split(".") + out: list[int] = [] + for p in parts: + if not p.isdigit(): + return None + out.append(int(p)) + return tuple(out) if out else None + def is_highest_version_tag(tag: str) -> bool: """ Return True if `tag` is the highest version among all tags matching v*. - Comparison uses `sort -V` for natural version ordering. + + We avoid shelling out to `sort -V` and implement a small vX.Y.Z parser. + Non-parseable v* tags are ignored for version comparison. """ - all_v = _capture("git tag --list 'v*'") + all_v = list_tags("v*") if not all_v: - return True # No tags yet, so the current tag is the highest + return True # No tags yet -> current is highest by definition - # Get the latest tag in natural version order - latest = _capture("git tag --list 'v*' | sort -V | tail -n1") - print(f"[INFO] Latest tag: {latest}, Current tag: {tag}") - - # Ensure that the current tag is always considered the highest if it's the latest one - return tag >= latest # Use comparison operator to consider all future tags + parsed_current = _parse_v_tag(tag) + if parsed_current is None: + # If the "current" tag isn't parseable, fall back to conservative behavior: + # treat it as highest only if it matches the max lexicographically. + latest_lex = max(all_v) + print(f"[INFO] Latest tag (lex): {latest_lex}, Current tag: {tag}") + return tag >= latest_lex + + parsed_all: list[tuple[int, ...]] = [] + for t in all_v: + parsed = _parse_v_tag(t) + if parsed is not None: + parsed_all.append(parsed) + + if not parsed_all: + # No parseable tags -> nothing to compare against + return True + + latest = max(parsed_all) + print(f"[INFO] Latest tag (parsed): v{'.'.join(map(str, latest))}, Current tag: {tag}") + return parsed_current >= latest -def update_latest_tag(new_tag: str, preview: bool = False) -> None: +def update_latest_tag(new_tag: str, *, preview: bool = False) -> None: """ Move the floating 'latest' tag to the newly created release tag. @@ -78,15 +95,10 @@ def update_latest_tag(new_tag: str, preview: bool = False) -> None: target_ref = f"{new_tag}^{{}}" print(f"[INFO] Updating 'latest' tag to point at {new_tag} (commit {target_ref})...") - if preview: - print( - f'[PREVIEW] Would run: git tag -f -a latest {target_ref} ' - f'-m "Floating latest tag for {new_tag}"' - ) - print("[PREVIEW] Would run: git push origin latest --force") - return - - run_git_command( - f'git tag -f -a latest {target_ref} -m "Floating latest tag for {new_tag}"' + tag_force_annotated( + name="latest", + target=target_ref, + message=f"Floating latest tag for {new_tag}", + preview=preview, ) - run_git_command("git push origin latest --force") + push("origin", "latest", force=True, preview=preview) diff --git a/src/pkgmgr/actions/release/workflow.py b/src/pkgmgr/actions/release/workflow.py index 3a381a2..d1aae33 100644 --- a/src/pkgmgr/actions/release/workflow.py +++ b/src/pkgmgr/actions/release/workflow.py @@ -6,6 +6,7 @@ from typing import Optional from pkgmgr.actions.branch import close_branch from pkgmgr.core.git import GitError +from pkgmgr.core.git.commands import add, commit, push, tag_annotated from pkgmgr.core.git.queries import get_current_branch from pkgmgr.core.repository.paths import resolve_repo_paths @@ -21,7 +22,6 @@ from .files import ( from .git_ops import ( ensure_clean_and_synced, is_highest_version_tag, - run_git_command, update_latest_tag, ) from .prompts import confirm_proceed_release, should_delete_branch @@ -126,12 +126,11 @@ def _release_impl( existing_files = [p for p in files_to_add if isinstance(p, str) and p and os.path.exists(p)] if preview: - for path in existing_files: - print(f"[PREVIEW] Would run: git add {path}") - print(f'[PREVIEW] Would run: git commit -am "{commit_msg}"') - print(f'[PREVIEW] Would run: git tag -a {new_tag} -m "{tag_msg}"') - print(f"[PREVIEW] Would run: git push origin {branch}") - print(f"[PREVIEW] Would run: git push origin {new_tag}") + add(existing_files, preview=True) + commit(commit_msg, all=True, preview=True) + tag_annotated(new_tag, tag_msg, preview=True) + push("origin", branch, preview=True) + push("origin", new_tag, preview=True) if is_highest_version_tag(new_tag): update_latest_tag(new_tag, preview=True) @@ -145,15 +144,13 @@ def _release_impl( print(f"[PREVIEW] Would ask whether to delete branch {branch} after release.") return - for path in existing_files: - run_git_command(f"git add {path}") - - run_git_command(f'git commit -am "{commit_msg}"') - run_git_command(f'git tag -a {new_tag} -m "{tag_msg}"') + add(existing_files, preview=False) + commit(commit_msg, all=True, preview=False) + tag_annotated(new_tag, tag_msg, preview=False) # Push branch and ONLY the newly created version tag (no --tags) - run_git_command(f"git push origin {branch}") - run_git_command(f"git push origin {new_tag}") + push("origin", branch, preview=False) + push("origin", new_tag, preview=False) # Update 'latest' only if this is the highest version tag try: diff --git a/src/pkgmgr/core/git/commands/__init__.py b/src/pkgmgr/core/git/commands/__init__.py index 0787e48..5bb73de 100644 --- a/src/pkgmgr/core/git/commands/__init__.py +++ b/src/pkgmgr/core/git/commands/__init__.py @@ -1,25 +1,33 @@ from __future__ import annotations +from .add import GitAddError, add from .checkout import GitCheckoutError, checkout +from .commit import GitCommitError, commit +from .create_branch import GitCreateBranchError, create_branch 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 .pull_ff_only import GitPullFfOnlyError, pull_ff_only 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 +from .set_remote_url import GitSetRemoteUrlError, set_remote_url +from .tag_annotated import GitTagAnnotatedError, tag_annotated +from .tag_force_annotated import GitTagForceAnnotatedError, tag_force_annotated __all__ = [ + "add", "fetch", "checkout", "pull", + "pull_ff_only", "merge_no_ff", "push", + "commit", "delete_local_branch", "delete_remote_branch", "create_branch", @@ -27,11 +35,16 @@ __all__ = [ "add_remote", "set_remote_url", "add_remote_push_url", + "tag_annotated", + "tag_force_annotated", + "GitAddError", "GitFetchError", "GitCheckoutError", "GitPullError", + "GitPullFfOnlyError", "GitMergeError", "GitPushError", + "GitCommitError", "GitDeleteLocalBranchError", "GitDeleteRemoteBranchError", "GitCreateBranchError", @@ -39,4 +52,6 @@ __all__ = [ "GitAddRemoteError", "GitSetRemoteUrlError", "GitAddRemotePushUrlError", + "GitTagAnnotatedError", + "GitTagForceAnnotatedError", ] diff --git a/src/pkgmgr/core/git/commands/add.py b/src/pkgmgr/core/git/commands/add.py new file mode 100644 index 0000000..d79bb4d --- /dev/null +++ b/src/pkgmgr/core/git/commands/add.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from typing import Iterable, List, Sequence, Union + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitAddError(GitCommandError): + """Raised when `git add` fails.""" + + +PathLike = Union[str, Sequence[str], Iterable[str]] + + +def _normalize_paths(paths: PathLike) -> List[str]: + if isinstance(paths, str): + return [paths] + return [p for p in paths] + + +def add( + paths: PathLike, + *, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Stage one or multiple paths. + + Equivalent to: + git add + """ + normalized = _normalize_paths(paths) + if not normalized: + return + + try: + run(["add", *normalized], cwd=cwd, preview=preview) + except GitError as exc: + raise GitAddError( + f"Failed to add paths to staging area: {normalized!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/commit.py b/src/pkgmgr/core/git/commands/commit.py new file mode 100644 index 0000000..73bb522 --- /dev/null +++ b/src/pkgmgr/core/git/commands/commit.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitCommitError(GitCommandError): + """Raised when `git commit` fails.""" + + +def commit( + message: str, + *, + cwd: str = ".", + all: bool = False, + preview: bool = False, +) -> None: + """ + Create a commit. + + Equivalent to: + git commit -m "" + or (if all=True): + git commit -am "" + """ + args = ["commit"] + if all: + args.append("-a") + args += ["-m", message] + + try: + run(args, cwd=cwd, preview=preview) + except GitError as exc: + raise GitCommitError( + "Failed to create commit.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/fetch.py b/src/pkgmgr/core/git/commands/fetch.py index 2b60d63..7e1d1a0 100644 --- a/src/pkgmgr/core/git/commands/fetch.py +++ b/src/pkgmgr/core/git/commands/fetch.py @@ -8,9 +8,31 @@ class GitFetchError(GitCommandError): """Raised when fetching from a remote fails.""" -def fetch(remote: str = "origin", cwd: str = ".") -> None: +def fetch( + remote: str = "origin", + *, + prune: bool = False, + tags: bool = False, + force: bool = False, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Fetch from a remote, optionally with prune/tags/force. + + Equivalent to: + git fetch [--prune] [--tags] [--force] + """ + args = ["fetch", remote] + if prune: + args.append("--prune") + if tags: + args.append("--tags") + if force: + args.append("--force") + try: - run(["fetch", remote], cwd=cwd) + run(args, cwd=cwd, preview=preview) except GitError as exc: raise GitFetchError( f"Failed to fetch from remote {remote!r}.", diff --git a/src/pkgmgr/core/git/commands/pull_ff_only.py b/src/pkgmgr/core/git/commands/pull_ff_only.py new file mode 100644 index 0000000..9ea1d2a --- /dev/null +++ b/src/pkgmgr/core/git/commands/pull_ff_only.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitPullFfOnlyError(GitCommandError): + """Raised when pulling with --ff-only fails.""" + + +def pull_ff_only(*, cwd: str = ".", preview: bool = False) -> None: + """ + Pull using fast-forward only. + + Equivalent to: + git pull --ff-only + """ + try: + run(["pull", "--ff-only"], cwd=cwd, preview=preview) + except GitError as exc: + raise GitPullFfOnlyError( + "Failed to pull with --ff-only.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/push.py b/src/pkgmgr/core/git/commands/push.py index bbe9485..1dc394c 100644 --- a/src/pkgmgr/core/git/commands/push.py +++ b/src/pkgmgr/core/git/commands/push.py @@ -8,9 +8,26 @@ class GitPushError(GitCommandError): """Raised when pushing to a remote fails.""" -def push(remote: str, ref: str, cwd: str = ".") -> None: +def push( + remote: str, + ref: str, + *, + force: bool = False, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Push a ref to a remote, optionally forced. + + Equivalent to: + git push [--force] + """ + args = ["push", remote, ref] + if force: + args.append("--force") + try: - run(["push", remote, ref], cwd=cwd) + run(args, cwd=cwd, preview=preview) except GitError as exc: raise GitPushError( f"Failed to push ref {ref!r} to remote {remote!r}.", diff --git a/src/pkgmgr/core/git/commands/tag_annotated.py b/src/pkgmgr/core/git/commands/tag_annotated.py new file mode 100644 index 0000000..5939a6f --- /dev/null +++ b/src/pkgmgr/core/git/commands/tag_annotated.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitTagAnnotatedError(GitCommandError): + """Raised when creating an annotated tag fails.""" + + +def tag_annotated( + tag: str, + message: str, + *, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Create an annotated tag. + + Equivalent to: + git tag -a -m "" + """ + try: + run(["tag", "-a", tag, "-m", message], cwd=cwd, preview=preview) + except GitError as exc: + raise GitTagAnnotatedError( + f"Failed to create annotated tag {tag!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/commands/tag_force_annotated.py b/src/pkgmgr/core/git/commands/tag_force_annotated.py new file mode 100644 index 0000000..aaeed58 --- /dev/null +++ b/src/pkgmgr/core/git/commands/tag_force_annotated.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitTagForceAnnotatedError(GitCommandError): + """Raised when forcing an annotated tag fails.""" + + +def tag_force_annotated( + name: str, + target: str, + message: str, + *, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Force-create an annotated tag pointing at a given target. + + Equivalent to: + git tag -f -a -m "" + """ + try: + run(["tag", "-f", "-a", name, target, "-m", message], cwd=cwd, preview=preview) + except GitError as exc: + raise GitTagForceAnnotatedError( + f"Failed to force annotated tag {name!r} at {target!r}.", + cwd=cwd, + ) from exc diff --git a/src/pkgmgr/core/git/queries/__init__.py b/src/pkgmgr/core/git/queries/__init__.py index d0890ff..6952657 100644 --- a/src/pkgmgr/core/git/queries/__init__.py +++ b/src/pkgmgr/core/git/queries/__init__.py @@ -11,6 +11,8 @@ 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 from .get_config_value import get_config_value +from .get_upstream_ref import get_upstream_ref +from .list_tags import list_tags __all__ = [ "get_current_branch", @@ -27,4 +29,6 @@ __all__ = [ "get_tags_at_ref", "GitTagsAtRefQueryError", "get_config_value", + "get_upstream_ref", + "list_tags", ] diff --git a/src/pkgmgr/core/git/queries/get_upstream_ref.py b/src/pkgmgr/core/git/queries/get_upstream_ref.py new file mode 100644 index 0000000..4416f3b --- /dev/null +++ b/src/pkgmgr/core/git/queries/get_upstream_ref.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Optional + +from ..errors import GitError +from ..run import run + + +def get_upstream_ref(*, cwd: str = ".") -> Optional[str]: + """ + Return the configured upstream ref for the current branch, or None if none. + + Equivalent to: + git rev-parse --abbrev-ref --symbolic-full-name @{u} + """ + try: + out = run( + ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"], + cwd=cwd, + ) + except GitError: + return None + + out = out.strip() + return out or None diff --git a/src/pkgmgr/core/git/queries/list_tags.py b/src/pkgmgr/core/git/queries/list_tags.py new file mode 100644 index 0000000..8d57c9d --- /dev/null +++ b/src/pkgmgr/core/git/queries/list_tags.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from typing import List + +from ..run import run + + +def list_tags(pattern: str = "*", *, cwd: str = ".") -> List[str]: + """ + List tags matching a pattern. + + Equivalent to: + git tag --list + """ + out = run(["tag", "--list", pattern], cwd=cwd) + if not out: + return [] + return [line.strip() for line in out.splitlines() if line.strip()] diff --git a/tests/unit/pkgmgr/actions/release/test_git_ops.py b/tests/unit/pkgmgr/actions/release/test_git_ops.py deleted file mode 100644 index a458841..0000000 --- a/tests/unit/pkgmgr/actions/release/test_git_ops.py +++ /dev/null @@ -1,198 +0,0 @@ -from __future__ import annotations - -import unittest -from unittest.mock import patch - -from pkgmgr.core.git import GitError -from pkgmgr.actions.release.git_ops import ( - ensure_clean_and_synced, - is_highest_version_tag, - run_git_command, - update_latest_tag, -) - - -class TestRunGitCommand(unittest.TestCase): - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_run_git_command_success(self, mock_run) -> None: - run_git_command("git status") - mock_run.assert_called_once() - args, kwargs = mock_run.call_args - self.assertIn("git status", args[0]) - self.assertTrue(kwargs.get("check")) - self.assertTrue(kwargs.get("capture_output")) - self.assertTrue(kwargs.get("text")) - - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_run_git_command_failure_raises_git_error(self, mock_run) -> None: - from subprocess import CalledProcessError - - mock_run.side_effect = CalledProcessError( - returncode=1, - cmd="git status", - output="stdout", - stderr="stderr", - ) - - with self.assertRaises(GitError): - run_git_command("git status") - - -class TestEnsureCleanAndSynced(unittest.TestCase): - def _fake_run(self, cmd: str, *args, **kwargs): - class R: - def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0): - self.stdout = stdout - self.stderr = stderr - self.returncode = returncode - - # upstream detection - if "git rev-parse --abbrev-ref --symbolic-full-name @{u}" in cmd: - return R(stdout="origin/main") - - # fetch/pull should be invoked in real mode - if cmd == "git fetch --prune --tags": - return R(stdout="") - if cmd == "git pull --ff-only": - return R(stdout="Already up to date.") - - return R(stdout="") - - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_ensure_clean_and_synced_preview_does_not_run_git_commands(self, mock_run) -> None: - def fake(cmd: str, *args, **kwargs): - class R: - def __init__(self, stdout: str = ""): - self.stdout = stdout - self.stderr = "" - self.returncode = 0 - - if "git rev-parse --abbrev-ref --symbolic-full-name @{u}" in cmd: - return R(stdout="origin/main") - return R(stdout="") - - mock_run.side_effect = fake - - ensure_clean_and_synced(preview=True) - - called_cmds = [c.args[0] for c in mock_run.call_args_list] - self.assertTrue(any("git rev-parse" in c for c in called_cmds)) - self.assertFalse(any(c == "git fetch --prune --tags" for c in called_cmds)) - self.assertFalse(any(c == "git pull --ff-only" for c in called_cmds)) - - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_ensure_clean_and_synced_no_upstream_skips(self, mock_run) -> None: - def fake(cmd: str, *args, **kwargs): - class R: - def __init__(self, stdout: str = ""): - self.stdout = stdout - self.stderr = "" - self.returncode = 0 - - if "git rev-parse --abbrev-ref --symbolic-full-name @{u}" in cmd: - return R(stdout="") # no upstream - return R(stdout="") - - mock_run.side_effect = fake - - ensure_clean_and_synced(preview=False) - - called_cmds = [c.args[0] for c in mock_run.call_args_list] - self.assertTrue(any("git rev-parse" in c for c in called_cmds)) - self.assertFalse(any(c == "git fetch --prune --tags" for c in called_cmds)) - self.assertFalse(any(c == "git pull --ff-only" for c in called_cmds)) - - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_ensure_clean_and_synced_real_runs_fetch_and_pull(self, mock_run) -> None: - mock_run.side_effect = self._fake_run - - ensure_clean_and_synced(preview=False) - - called_cmds = [c.args[0] for c in mock_run.call_args_list] - self.assertIn("git fetch origin --prune --tags --force", called_cmds) - self.assertIn("git pull --ff-only", called_cmds) - - - -class TestIsHighestVersionTag(unittest.TestCase): - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_is_highest_version_tag_no_tags_true(self, mock_run) -> None: - def fake(cmd: str, *args, **kwargs): - class R: - def __init__(self, stdout: str = ""): - self.stdout = stdout - self.stderr = "" - self.returncode = 0 - - if "git tag --list" in cmd and "'v*'" in cmd: - return R(stdout="") # no tags - return R(stdout="") - - mock_run.side_effect = fake - - self.assertTrue(is_highest_version_tag("v1.0.0")) - - # ensure at least the list command was queried - called_cmds = [c.args[0] for c in mock_run.call_args_list] - self.assertTrue(any("git tag --list" in c for c in called_cmds)) - - @patch("pkgmgr.actions.release.git_ops.subprocess.run") - def test_is_highest_version_tag_compares_sort_v(self, mock_run) -> None: - """ - This test is aligned with the CURRENT implementation: - - return tag >= latest - - which is a *string comparison*, not a semantic version compare. - Therefore, a candidate like v1.2.0 is lexicographically >= v1.10.0 - (because '2' > '1' at the first differing char after 'v1.'). - """ - def fake(cmd: str, *args, **kwargs): - class R: - def __init__(self, stdout: str = ""): - self.stdout = stdout - self.stderr = "" - self.returncode = 0 - - if cmd.strip() == "git tag --list 'v*'": - return R(stdout="v1.0.0\nv1.2.0\nv1.10.0\n") - if "git tag --list 'v*'" in cmd and "sort -V" in cmd and "tail -n1" in cmd: - return R(stdout="v1.10.0") - return R(stdout="") - - mock_run.side_effect = fake - - # With the current implementation (string >=), both of these are True. - self.assertTrue(is_highest_version_tag("v1.10.0")) - self.assertTrue(is_highest_version_tag("v1.2.0")) - - # And a clearly lexicographically smaller candidate should be False. - # Example: "v1.0.0" < "v1.10.0" - self.assertFalse(is_highest_version_tag("v1.0.0")) - - # Ensure both capture commands were executed - called_cmds = [c.args[0] for c in mock_run.call_args_list] - self.assertTrue(any(cmd == "git tag --list 'v*'" for cmd in called_cmds)) - self.assertTrue(any("sort -V" in cmd and "tail -n1" in cmd for cmd in called_cmds)) - - -class TestUpdateLatestTag(unittest.TestCase): - @patch("pkgmgr.actions.release.git_ops.run_git_command") - def test_update_latest_tag_preview_does_not_call_git(self, mock_run_git_command) -> None: - update_latest_tag("v1.2.3", preview=True) - mock_run_git_command.assert_not_called() - - @patch("pkgmgr.actions.release.git_ops.run_git_command") - def test_update_latest_tag_real_calls_git(self, mock_run_git_command) -> None: - update_latest_tag("v1.2.3", preview=False) - - calls = [c.args[0] for c in mock_run_git_command.call_args_list] - self.assertIn( - 'git tag -f -a latest v1.2.3^{} -m "Floating latest tag for v1.2.3"', - calls, - ) - self.assertIn("git push origin latest --force", calls) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit/pkgmgr/actions/release/test_git_ops_ensure_clean_and_synced.py b/tests/unit/pkgmgr/actions/release/test_git_ops_ensure_clean_and_synced.py new file mode 100644 index 0000000..e9bf75c --- /dev/null +++ b/tests/unit/pkgmgr/actions/release/test_git_ops_ensure_clean_and_synced.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from pkgmgr.actions.release.git_ops import ensure_clean_and_synced + + +class TestEnsureCleanAndSynced(unittest.TestCase): + @patch("pkgmgr.actions.release.git_ops.pull_ff_only") + @patch("pkgmgr.actions.release.git_ops.fetch") + @patch("pkgmgr.actions.release.git_ops.get_upstream_ref") + def test_no_upstream_skips( + self, + mock_get_upstream_ref, + mock_fetch, + mock_pull_ff_only, + ) -> None: + mock_get_upstream_ref.return_value = None + + ensure_clean_and_synced(preview=False) + + mock_fetch.assert_not_called() + mock_pull_ff_only.assert_not_called() + + @patch("pkgmgr.actions.release.git_ops.pull_ff_only") + @patch("pkgmgr.actions.release.git_ops.fetch") + @patch("pkgmgr.actions.release.git_ops.get_upstream_ref") + def test_preview_calls_commands_with_preview_true( + self, + mock_get_upstream_ref, + mock_fetch, + mock_pull_ff_only, + ) -> None: + mock_get_upstream_ref.return_value = "origin/main" + + ensure_clean_and_synced(preview=True) + + mock_fetch.assert_called_once_with( + remote="origin", + prune=True, + tags=True, + force=True, + preview=True, + ) + mock_pull_ff_only.assert_called_once_with(preview=True) + + @patch("pkgmgr.actions.release.git_ops.pull_ff_only") + @patch("pkgmgr.actions.release.git_ops.fetch") + @patch("pkgmgr.actions.release.git_ops.get_upstream_ref") + def test_real_calls_commands_with_preview_false( + self, + mock_get_upstream_ref, + mock_fetch, + mock_pull_ff_only, + ) -> None: + mock_get_upstream_ref.return_value = "origin/main" + + ensure_clean_and_synced(preview=False) + + mock_fetch.assert_called_once_with( + remote="origin", + prune=True, + tags=True, + force=True, + preview=False, + ) + mock_pull_ff_only.assert_called_once_with(preview=False) diff --git a/tests/unit/pkgmgr/actions/release/test_git_ops_is_highest_version_tag.py b/tests/unit/pkgmgr/actions/release/test_git_ops_is_highest_version_tag.py new file mode 100644 index 0000000..fab6ce1 --- /dev/null +++ b/tests/unit/pkgmgr/actions/release/test_git_ops_is_highest_version_tag.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from pkgmgr.actions.release.git_ops import is_highest_version_tag + + +class TestIsHighestVersionTag(unittest.TestCase): + @patch("pkgmgr.actions.release.git_ops.list_tags") + def test_no_tags_returns_true(self, mock_list_tags) -> None: + mock_list_tags.return_value = [] + self.assertTrue(is_highest_version_tag("v1.0.0")) + mock_list_tags.assert_called_once_with("v*") + + @patch("pkgmgr.actions.release.git_ops.list_tags") + def test_parseable_semver_compares_correctly(self, mock_list_tags) -> None: + # Highest is v1.10.0 (semantic compare) + mock_list_tags.return_value = ["v1.0.0", "v1.2.0", "v1.10.0"] + + self.assertTrue(is_highest_version_tag("v1.10.0")) + self.assertFalse(is_highest_version_tag("v1.2.0")) + self.assertFalse(is_highest_version_tag("v1.0.0")) + + @patch("pkgmgr.actions.release.git_ops.list_tags") + def test_ignores_non_parseable_v_tags_for_semver_compare(self, mock_list_tags) -> None: + mock_list_tags.return_value = ["v1.2.0", "v1.10.0", "v1.2.0-rc1", "vfoo"] + + self.assertTrue(is_highest_version_tag("v1.10.0")) + self.assertFalse(is_highest_version_tag("v1.2.0")) + + @patch("pkgmgr.actions.release.git_ops.list_tags") + def test_current_tag_not_parseable_falls_back_to_lex_compare(self, mock_list_tags) -> None: + mock_list_tags.return_value = ["v1.9.0", "v1.10.0"] + + # prerelease must NOT outrank the final release + self.assertFalse(is_highest_version_tag("v1.10.0-rc1")) + self.assertFalse(is_highest_version_tag("v1.0.0-rc1")) + + diff --git a/tests/unit/pkgmgr/actions/release/test_git_ops_update_latest_tag.py b/tests/unit/pkgmgr/actions/release/test_git_ops_update_latest_tag.py new file mode 100644 index 0000000..3abccdd --- /dev/null +++ b/tests/unit/pkgmgr/actions/release/test_git_ops_update_latest_tag.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from pkgmgr.actions.release.git_ops import update_latest_tag + + +class TestUpdateLatestTag(unittest.TestCase): + @patch("pkgmgr.actions.release.git_ops.push") + @patch("pkgmgr.actions.release.git_ops.tag_force_annotated") + def test_preview_calls_commands_with_preview_true( + self, + mock_tag_force_annotated, + mock_push, + ) -> None: + update_latest_tag("v1.2.3", preview=True) + + mock_tag_force_annotated.assert_called_once_with( + name="latest", + target="v1.2.3^{}", + message="Floating latest tag for v1.2.3", + preview=True, + ) + mock_push.assert_called_once_with( + "origin", + "latest", + force=True, + preview=True, + ) + + @patch("pkgmgr.actions.release.git_ops.push") + @patch("pkgmgr.actions.release.git_ops.tag_force_annotated") + def test_real_calls_commands_with_preview_false( + self, + mock_tag_force_annotated, + mock_push, + ) -> None: + update_latest_tag("v1.2.3", preview=False) + + mock_tag_force_annotated.assert_called_once_with( + name="latest", + target="v1.2.3^{}", + message="Floating latest tag for v1.2.3", + preview=False, + ) + mock_push.assert_called_once_with( + "origin", + "latest", + force=True, + preview=False, + ) diff --git a/tests/unit/pkgmgr/core/git/test_run.py b/tests/unit/pkgmgr/core/git/test_run.py new file mode 100644 index 0000000..eeaadfc --- /dev/null +++ b/tests/unit/pkgmgr/core/git/test_run.py @@ -0,0 +1,69 @@ +import unittest +from unittest.mock import MagicMock, patch + +from pkgmgr.core.git.errors import GitError +from pkgmgr.core.git.run import run + + +class TestGitRun(unittest.TestCase): + def test_preview_mode_prints_and_does_not_execute(self) -> None: + with patch("pkgmgr.core.git.run.subprocess.run") as mock_run, patch( + "builtins.print" + ) as mock_print: + out = run(["status"], cwd="/tmp/repo", preview=True) + + self.assertEqual(out, "") + mock_run.assert_not_called() + mock_print.assert_called_once() + printed = mock_print.call_args[0][0] + self.assertIn("[PREVIEW] Would run in '/tmp/repo': git status", printed) + + def test_success_returns_stripped_stdout(self) -> None: + completed = MagicMock() + completed.stdout = " hello world \n" + completed.stderr = "" + completed.returncode = 0 + + with patch("pkgmgr.core.git.run.subprocess.run", return_value=completed) as mock_run: + out = run(["rev-parse", "HEAD"], cwd="/repo", preview=False) + + self.assertEqual(out, "hello world") + + mock_run.assert_called_once() + args, kwargs = mock_run.call_args + self.assertEqual(args[0], ["git", "rev-parse", "HEAD"]) + self.assertEqual(kwargs["cwd"], "/repo") + self.assertTrue(kwargs["check"]) + self.assertTrue(kwargs["text"]) + # ensure pipes are used (matches implementation intent) + self.assertIsNotNone(kwargs["stdout"]) + self.assertIsNotNone(kwargs["stderr"]) + + def test_failure_raises_giterror_with_details(self) -> None: + # Build a CalledProcessError with stdout/stderr populated + import subprocess as sp + + exc = sp.CalledProcessError( + returncode=128, + cmd=["git", "status"], + output="OUT!", + stderr="ERR!", + ) + # Your implementation reads exc.stdout, but CalledProcessError stores it as .output + # in some cases. Ensure .stdout exists for deterministic behavior. + exc.stdout = "OUT!" + exc.stderr = "ERR!" + + with patch("pkgmgr.core.git.run.subprocess.run", side_effect=exc): + with self.assertRaises(GitError) as ctx: + run(["status"], cwd="/bad/repo", preview=False) + + msg = str(ctx.exception) + self.assertIn("Git command failed in '/bad/repo': git status", msg) + self.assertIn("Exit code: 128", msg) + self.assertIn("STDOUT:\nOUT!", msg) + self.assertIn("STDERR:\nERR!", msg) + + +if __name__ == "__main__": + unittest.main()