refactor(release/git): replace shell git calls with command/query helpers
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-env-virtual (push) Has been cancelled
Mark stable commit / test-env-nix (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / lint-shell (push) Has been cancelled
Mark stable commit / lint-python (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-env-virtual (push) Has been cancelled
Mark stable commit / test-env-nix (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / lint-shell (push) Has been cancelled
Mark stable commit / lint-python (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
- Remove legacy shell-based git helpers from release workflow - Introduce typed git command wrappers (add, commit, fetch, pull_ff_only, push, tag*) - Add git queries for upstream detection and tag listing - Refactor release workflow to use core git commands consistently - Implement semantic vX.Y.Z tag comparison without external sort - Ensure prerelease tags (e.g. -rc) do not outrank final releases - Split and update unit tests to match new command/query architecture
This commit is contained in:
@@ -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()
|
||||
@@ -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)
|
||||
@@ -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"))
|
||||
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
69
tests/unit/pkgmgr/core/git/test_run.py
Normal file
69
tests/unit/pkgmgr/core/git/test_run.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user