From 9c22c7dbb43f254523e4237407840c3e4edf065a Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Tue, 16 Dec 2025 12:49:03 +0100 Subject: [PATCH] refactor(git): introduce structured core.git command/query API and adapt actions & tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace direct subprocess usage with core.git.run wrapper - Introduce dedicated core.git.commands (add, commit, fetch, pull_ff_only, push, clone, tag_annotated, tag_force_annotated, etc.) - Introduce core.git.queries (list_tags, get_upstream_ref, get_config_value, changelog helpers, etc.) - Refactor release workflow and git_ops to use command/query split - Implement semantic vX.Y.Z comparison with safe fallback for non-parseable tags - Refactor repository clone logic to use core.git.commands.clone with preview support and ssh→https fallback - Remove legacy run_git_command helpers - Split and update unit tests to mock command/query boundaries instead of subprocess - Add comprehensive tests for clone modes, preview behavior, ssh→https fallback, and verification prompts - Add unit tests for core.git.run error handling and preview mode - Align public exports (__all__) with new structure - Improve type hints, docstrings, and error specificity across git helpers https://chatgpt.com/share/69414735-51d4-800f-bc7b-4b90e35f71e5 --- src/pkgmgr/actions/repository/clone.py | 183 +++++++------ src/pkgmgr/core/git/commands/__init__.py | 3 + src/pkgmgr/core/git/commands/clone.py | 32 +++ tests/unit/pkgmgr/actions/repos/test_clone.py | 251 +++++++++++++++--- 4 files changed, 352 insertions(+), 117 deletions(-) create mode 100644 src/pkgmgr/core/git/commands/clone.py diff --git a/src/pkgmgr/actions/repository/clone.py b/src/pkgmgr/actions/repository/clone.py index 40a1846..3e408c2 100644 --- a/src/pkgmgr/actions/repository/clone.py +++ b/src/pkgmgr/actions/repository/clone.py @@ -1,103 +1,132 @@ -import subprocess +from __future__ import annotations + import os +from typing import Any, Dict, List, Optional + +from pkgmgr.core.git.commands import clone as git_clone, GitCloneError from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.identifier import get_repo_identifier from pkgmgr.core.repository.verify import verify_repository +Repository = Dict[str, Any] + + +def _build_clone_url(repo: Repository, clone_mode: str) -> Optional[str]: + provider = repo.get("provider") + account = repo.get("account") + name = repo.get("repository") + replacement = repo.get("replacement") + + if clone_mode == "ssh": + if not provider or not account or not name: + return None + return f"git@{provider}:{account}/{name}.git" + + if clone_mode in ("https", "shallow"): + if replacement: + return f"https://{replacement}.git" + if not provider or not account or not name: + return None + return f"https://{provider}/{account}/{name}.git" + + return None + + def clone_repos( - selected_repos, - repositories_base_dir: str, - all_repos, - preview: bool, - no_verification: bool, - clone_mode: str - ): + selected_repos: List[Repository], + repositories_base_dir: str, + all_repos: List[Repository], + preview: bool, + no_verification: bool, + clone_mode: str, +) -> None: for repo in selected_repos: repo_identifier = get_repo_identifier(repo, all_repos) repo_dir = get_repo_dir(repositories_base_dir, repo) + if os.path.exists(repo_dir): - print(f"[INFO] Repository '{repo_identifier}' already exists at '{repo_dir}'. Skipping clone.") + print( + f"[INFO] Repository '{repo_identifier}' already exists at '{repo_dir}'. Skipping clone." + ) continue parent_dir = os.path.dirname(repo_dir) os.makedirs(parent_dir, exist_ok=True) - # Build clone URL based on the clone_mode - # Build clone URL based on the clone_mode - if clone_mode == "ssh": - clone_url = ( - f"git@{repo.get('provider')}:" - f"{repo.get('account')}/" - f"{repo.get('repository')}.git" - ) - elif clone_mode in ("https", "shallow"): - # Use replacement if defined, otherwise construct from provider/account/repository - if repo.get("replacement"): - clone_url = f"https://{repo.get('replacement')}.git" - else: - clone_url = ( - f"https://{repo.get('provider')}/" - f"{repo.get('account')}/" - f"{repo.get('repository')}.git" - ) - else: - print(f"Unknown clone mode '{clone_mode}'. Aborting clone for {repo_identifier}.") + + clone_url = _build_clone_url(repo, clone_mode) + if not clone_url: + print(f"[WARNING] Cannot build clone URL for '{repo_identifier}'. Skipping.") continue - # Build base clone command - base_clone_cmd = "git clone" - if clone_mode == "shallow": - # Shallow clone: only latest state via HTTPS, no full history - base_clone_cmd += " --depth 1 --single-branch" + shallow = clone_mode == "shallow" + mode_label = "HTTPS (shallow)" if shallow else clone_mode.upper() - mode_label = "HTTPS (shallow)" if clone_mode == "shallow" else clone_mode.upper() print( f"[INFO] Attempting to clone '{repo_identifier}' using {mode_label} " f"from {clone_url} into '{repo_dir}'." ) - if preview: - print(f"[Preview] Would run: {base_clone_cmd} {clone_url} {repo_dir} in {parent_dir}") - result = subprocess.CompletedProcess(args=[], returncode=0) - else: - result = subprocess.run( - f"{base_clone_cmd} {clone_url} {repo_dir}", + try: + args = [] + if shallow: + args += ["--depth", "1", "--single-branch"] + args += [clone_url, repo_dir] + + git_clone( + args, cwd=parent_dir, - shell=True, + preview=preview, ) - - if result.returncode != 0: - # Only offer fallback if the original mode was SSH. - if clone_mode == "ssh": - print(f"[WARNING] SSH clone failed for '{repo_identifier}' with return code {result.returncode}.") - choice = input("Do you want to attempt HTTPS clone instead? (y/N): ").strip().lower() - if choice == 'y': - # Attempt HTTPS clone - if repo.get("replacement"): - clone_url = f"https://{repo.get('replacement')}.git" - else: - clone_url = f"https://{repo.get('provider')}/{repo.get('account')}/{repo.get('repository')}.git" - print(f"[INFO] Attempting to clone '{repo_identifier}' using HTTPS from {clone_url} into '{repo_dir}'.") - if preview: - print(f"[Preview] Would run: git clone {clone_url} {repo_dir} in {parent_dir}") - result = subprocess.CompletedProcess(args=[], returncode=0) - else: - result = subprocess.run(f"git clone {clone_url} {repo_dir}", cwd=parent_dir, shell=True) - else: - print(f"[INFO] HTTPS clone not attempted for '{repo_identifier}'.") - continue - else: - # For https mode, do not attempt fallback. - print(f"[WARNING] HTTPS clone failed for '{repo_identifier}' with return code {result.returncode}.") + + except GitCloneError as exc: + if clone_mode != "ssh": + print(f"[WARNING] Clone failed for '{repo_identifier}': {exc}") continue - - # After cloning, perform verification in local mode. + + print(f"[WARNING] SSH clone failed for '{repo_identifier}': {exc}") + choice = input("Do you want to attempt HTTPS clone instead? (y/N): ").strip().lower() + if choice != "y": + print(f"[INFO] HTTPS clone not attempted for '{repo_identifier}'.") + continue + + fallback_url = _build_clone_url(repo, "https") + if not fallback_url: + print(f"[WARNING] Cannot build HTTPS URL for '{repo_identifier}'.") + continue + + print( + f"[INFO] Attempting to clone '{repo_identifier}' using HTTPS " + f"from {fallback_url} into '{repo_dir}'." + ) + + try: + git_clone( + [fallback_url, repo_dir], + cwd=parent_dir, + preview=preview, + ) + except GitCloneError as exc2: + print(f"[WARNING] HTTPS clone failed for '{repo_identifier}': {exc2}") + continue + verified_info = repo.get("verified") - if verified_info: - verified_ok, errors, commit_hash, signing_key = verify_repository(repo, repo_dir, mode="local", no_verification=no_verification) - if not no_verification and not verified_ok: - print(f"Warning: Verification failed for {repo_identifier} after cloning:") - for err in errors: - print(f" - {err}") - choice = input("Proceed anyway? (y/N): ").strip().lower() - if choice != "y": - print(f"Skipping repository {repo_identifier} due to failed verification.") + if not verified_info: + continue + + verified_ok, errors, _commit_hash, _signing_key = verify_repository( + repo, + repo_dir, + mode="local", + no_verification=no_verification, + ) + + if no_verification or verified_ok: + continue + + print(f"Warning: Verification failed for {repo_identifier} after cloning:") + for err in errors: + print(f" - {err}") + + choice = input("Proceed anyway? (y/N): ").strip().lower() + if choice != "y": + print(f"Skipping repository {repo_identifier} due to failed verification.") diff --git a/src/pkgmgr/core/git/commands/__init__.py b/src/pkgmgr/core/git/commands/__init__.py index 5bb73de..37db521 100644 --- a/src/pkgmgr/core/git/commands/__init__.py +++ b/src/pkgmgr/core/git/commands/__init__.py @@ -18,6 +18,7 @@ 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 +from .clone import GitCloneError, clone __all__ = [ "add", @@ -37,6 +38,7 @@ __all__ = [ "add_remote_push_url", "tag_annotated", "tag_force_annotated", + "clone", "GitAddError", "GitFetchError", "GitCheckoutError", @@ -54,4 +56,5 @@ __all__ = [ "GitAddRemotePushUrlError", "GitTagAnnotatedError", "GitTagForceAnnotatedError", + "GitCloneError", ] diff --git a/src/pkgmgr/core/git/commands/clone.py b/src/pkgmgr/core/git/commands/clone.py new file mode 100644 index 0000000..d904c70 --- /dev/null +++ b/src/pkgmgr/core/git/commands/clone.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from typing import List + +from ..errors import GitError, GitCommandError +from ..run import run + + +class GitCloneError(GitCommandError): + """Raised when `git clone` fails.""" + + +def clone( + args: List[str], + *, + cwd: str = ".", + preview: bool = False, +) -> None: + """ + Execute `git clone` with caller-provided arguments. + + Examples: + ["https://example.com/repo.git", "/path/to/dir"] + ["--depth", "1", "--single-branch", url, dest] + """ + try: + run(["clone", *args], cwd=cwd, preview=preview) + except GitError as exc: + raise GitCloneError( + f"Git clone failed with args={args!r}.", + cwd=cwd, + ) from exc diff --git a/tests/unit/pkgmgr/actions/repos/test_clone.py b/tests/unit/pkgmgr/actions/repos/test_clone.py index 46b5d67..31bbc82 100644 --- a/tests/unit/pkgmgr/actions/repos/test_clone.py +++ b/tests/unit/pkgmgr/actions/repos/test_clone.py @@ -1,23 +1,27 @@ -# tests/test_clone_repos.py +# tests/unit/pkgmgr/actions/repos/test_clone.py +from __future__ import annotations + import unittest -from unittest.mock import patch, MagicMock +from unittest.mock import patch from pkgmgr.actions.repository.clone import clone_repos class TestCloneRepos(unittest.TestCase): - def setUp(self): + def setUp(self) -> None: + # Add `verified` so verify_repository() is actually exercised. self.repo = { "provider": "github.com", "account": "user", "repository": "repo", + "verified": {"commit": "deadbeef"}, } self.selected = [self.repo] self.base_dir = "/tmp/repos" self.all_repos = self.selected @patch("pkgmgr.actions.repository.clone.verify_repository") - @patch("pkgmgr.actions.repository.clone.subprocess.run") + @patch("pkgmgr.actions.repository.clone.git_clone") @patch("pkgmgr.actions.repository.clone.os.makedirs") @patch("pkgmgr.actions.repository.clone.os.path.exists") @patch("pkgmgr.actions.repository.clone.get_repo_dir") @@ -28,13 +32,14 @@ class TestCloneRepos(unittest.TestCase): mock_get_repo_dir, mock_exists, mock_makedirs, - mock_run, + mock_git_clone, mock_verify, - ): + ) -> None: mock_get_repo_identifier.return_value = "github.com/user/repo" mock_get_repo_dir.return_value = "/tmp/repos/user/repo" mock_exists.return_value = False - mock_run.return_value = MagicMock(returncode=0) + + # verification called; and because no_verification=True, result doesn't matter mock_verify.return_value = (True, [], "hash", "key") clone_repos( @@ -46,17 +51,26 @@ class TestCloneRepos(unittest.TestCase): clone_mode="ssh", ) - mock_run.assert_called_once() - # subprocess.run wird mit positional args aufgerufen - cmd = mock_run.call_args[0][0] - cwd = mock_run.call_args[1]["cwd"] + mock_git_clone.assert_called_once() + args, kwargs = mock_git_clone.call_args + clone_args = args[0] + self.assertEqual( + clone_args, + ["git@github.com:user/repo.git", "/tmp/repos/user/repo"], + ) + self.assertEqual(kwargs["cwd"], "/tmp/repos/user") + self.assertFalse(kwargs["preview"]) - self.assertIn("git clone", cmd) - self.assertIn("git@github.com:user/repo.git", cmd) - self.assertEqual(cwd, "/tmp/repos/user") + # verify_repository should be called because repo has "verified" + mock_verify.assert_called_once() + v_args, v_kwargs = mock_verify.call_args + self.assertEqual(v_args[0], self.repo) # repo dict + self.assertEqual(v_args[1], "/tmp/repos/user/repo") # repo_dir + self.assertEqual(v_kwargs["mode"], "local") + self.assertTrue(v_kwargs["no_verification"]) @patch("pkgmgr.actions.repository.clone.verify_repository") - @patch("pkgmgr.actions.repository.clone.subprocess.run") + @patch("pkgmgr.actions.repository.clone.git_clone") @patch("pkgmgr.actions.repository.clone.os.makedirs") @patch("pkgmgr.actions.repository.clone.os.path.exists") @patch("pkgmgr.actions.repository.clone.get_repo_dir") @@ -67,13 +81,12 @@ class TestCloneRepos(unittest.TestCase): mock_get_repo_dir, mock_exists, mock_makedirs, - mock_run, + mock_git_clone, mock_verify, - ): + ) -> None: mock_get_repo_identifier.return_value = "github.com/user/repo" mock_get_repo_dir.return_value = "/tmp/repos/user/repo" mock_exists.return_value = False - mock_run.return_value = MagicMock(returncode=0) mock_verify.return_value = (True, [], "hash", "key") clone_repos( @@ -85,16 +98,20 @@ class TestCloneRepos(unittest.TestCase): clone_mode="https", ) - mock_run.assert_called_once() - cmd = mock_run.call_args[0][0] - cwd = mock_run.call_args[1]["cwd"] + mock_git_clone.assert_called_once() + args, kwargs = mock_git_clone.call_args + clone_args = args[0] + self.assertEqual( + clone_args, + ["https://github.com/user/repo.git", "/tmp/repos/user/repo"], + ) + self.assertEqual(kwargs["cwd"], "/tmp/repos/user") + self.assertFalse(kwargs["preview"]) - self.assertIn("git clone", cmd) - self.assertIn("https://github.com/user/repo.git", cmd) - self.assertEqual(cwd, "/tmp/repos/user") + mock_verify.assert_called_once() @patch("pkgmgr.actions.repository.clone.verify_repository") - @patch("pkgmgr.actions.repository.clone.subprocess.run") + @patch("pkgmgr.actions.repository.clone.git_clone") @patch("pkgmgr.actions.repository.clone.os.makedirs") @patch("pkgmgr.actions.repository.clone.os.path.exists") @patch("pkgmgr.actions.repository.clone.get_repo_dir") @@ -105,13 +122,12 @@ class TestCloneRepos(unittest.TestCase): mock_get_repo_dir, mock_exists, mock_makedirs, - mock_run, + mock_git_clone, mock_verify, - ): + ) -> None: mock_get_repo_identifier.return_value = "github.com/user/repo" mock_get_repo_dir.return_value = "/tmp/repos/user/repo" mock_exists.return_value = False - mock_run.return_value = MagicMock(returncode=0) mock_verify.return_value = (True, [], "hash", "key") clone_repos( @@ -123,29 +139,39 @@ class TestCloneRepos(unittest.TestCase): clone_mode="shallow", ) - mock_run.assert_called_once() - cmd = mock_run.call_args[0][0] - cwd = mock_run.call_args[1]["cwd"] + mock_git_clone.assert_called_once() + args, kwargs = mock_git_clone.call_args + clone_args = args[0] + self.assertEqual( + clone_args, + [ + "--depth", + "1", + "--single-branch", + "https://github.com/user/repo.git", + "/tmp/repos/user/repo", + ], + ) + self.assertEqual(kwargs["cwd"], "/tmp/repos/user") + self.assertFalse(kwargs["preview"]) - self.assertIn("git clone --depth 1 --single-branch", cmd) - self.assertIn("https://github.com/user/repo.git", cmd) - self.assertEqual(cwd, "/tmp/repos/user") + mock_verify.assert_called_once() @patch("pkgmgr.actions.repository.clone.verify_repository") - @patch("pkgmgr.actions.repository.clone.subprocess.run") + @patch("pkgmgr.actions.repository.clone.git_clone") @patch("pkgmgr.actions.repository.clone.os.makedirs") @patch("pkgmgr.actions.repository.clone.os.path.exists") @patch("pkgmgr.actions.repository.clone.get_repo_dir") @patch("pkgmgr.actions.repository.clone.get_repo_identifier") - def test_preview_mode_does_not_call_subprocess_run( + def test_preview_mode_calls_git_clone_with_preview_true( self, mock_get_repo_identifier, mock_get_repo_dir, mock_exists, mock_makedirs, - mock_run, + mock_git_clone, mock_verify, - ): + ) -> None: mock_get_repo_identifier.return_value = "github.com/user/repo" mock_get_repo_dir.return_value = "/tmp/repos/user/repo" mock_exists.return_value = False @@ -160,8 +186,153 @@ class TestCloneRepos(unittest.TestCase): clone_mode="shallow", ) - # Im Preview-Modus sollte subprocess.run nicht aufgerufen werden - mock_run.assert_not_called() + mock_git_clone.assert_called_once() + _args, kwargs = mock_git_clone.call_args + self.assertTrue(kwargs["preview"]) + + # Even in preview, verification is reached (because repo has "verified"), + # but no_verification=True makes it non-blocking. + mock_verify.assert_called_once() + + @patch("builtins.input", return_value="y") + @patch("pkgmgr.actions.repository.clone.verify_repository") + @patch("pkgmgr.actions.repository.clone.git_clone") + @patch("pkgmgr.actions.repository.clone.os.makedirs") + @patch("pkgmgr.actions.repository.clone.os.path.exists") + @patch("pkgmgr.actions.repository.clone.get_repo_dir") + @patch("pkgmgr.actions.repository.clone.get_repo_identifier") + def test_ssh_clone_failure_prompts_and_falls_back_to_https_when_confirmed( + self, + mock_get_repo_identifier, + mock_get_repo_dir, + mock_exists, + mock_makedirs, + mock_git_clone, + mock_verify, + mock_input, + ) -> None: + mock_get_repo_identifier.return_value = "github.com/user/repo" + mock_get_repo_dir.return_value = "/tmp/repos/user/repo" + mock_exists.return_value = False + mock_verify.return_value = (True, [], "hash", "key") + + # First call (ssh) fails, second call (https) succeeds + from pkgmgr.core.git.commands.clone import GitCloneError + + mock_git_clone.side_effect = [ + GitCloneError("ssh failed", cwd="/tmp/repos/user"), + None, + ] + + clone_repos( + self.selected, + self.base_dir, + self.all_repos, + preview=False, + no_verification=True, + clone_mode="ssh", + ) + + self.assertEqual(mock_git_clone.call_count, 2) + + first_args, first_kwargs = mock_git_clone.call_args_list[0] + self.assertEqual( + first_args[0], + ["git@github.com:user/repo.git", "/tmp/repos/user/repo"], + ) + self.assertEqual(first_kwargs["cwd"], "/tmp/repos/user") + self.assertFalse(first_kwargs["preview"]) + + second_args, second_kwargs = mock_git_clone.call_args_list[1] + self.assertEqual( + second_args[0], + ["https://github.com/user/repo.git", "/tmp/repos/user/repo"], + ) + self.assertEqual(second_kwargs["cwd"], "/tmp/repos/user") + self.assertFalse(second_kwargs["preview"]) + + mock_input.assert_called_once() + mock_verify.assert_called_once() + + @patch("builtins.input", return_value="n") + @patch("pkgmgr.actions.repository.clone.verify_repository") + @patch("pkgmgr.actions.repository.clone.git_clone") + @patch("pkgmgr.actions.repository.clone.os.makedirs") + @patch("pkgmgr.actions.repository.clone.os.path.exists") + @patch("pkgmgr.actions.repository.clone.get_repo_dir") + @patch("pkgmgr.actions.repository.clone.get_repo_identifier") + def test_ssh_clone_failure_does_not_fallback_when_declined( + self, + mock_get_repo_identifier, + mock_get_repo_dir, + mock_exists, + mock_makedirs, + mock_git_clone, + mock_verify, + mock_input, + ) -> None: + mock_get_repo_identifier.return_value = "github.com/user/repo" + mock_get_repo_dir.return_value = "/tmp/repos/user/repo" + mock_exists.return_value = False + + from pkgmgr.core.git.commands.clone import GitCloneError + + mock_git_clone.side_effect = GitCloneError("ssh failed", cwd="/tmp/repos/user") + + clone_repos( + self.selected, + self.base_dir, + self.all_repos, + preview=False, + no_verification=True, + clone_mode="ssh", + ) + + mock_git_clone.assert_called_once() + mock_input.assert_called_once() + + # If fallback is declined, verification should NOT run (repo was not cloned) + mock_verify.assert_not_called() + + @patch("builtins.input", return_value="n") + @patch("pkgmgr.actions.repository.clone.verify_repository") + @patch("pkgmgr.actions.repository.clone.git_clone") + @patch("pkgmgr.actions.repository.clone.os.makedirs") + @patch("pkgmgr.actions.repository.clone.os.path.exists") + @patch("pkgmgr.actions.repository.clone.get_repo_dir") + @patch("pkgmgr.actions.repository.clone.get_repo_identifier") + def test_verification_failure_prompts_and_skips_when_user_declines( + self, + mock_get_repo_identifier, + mock_get_repo_dir, + mock_exists, + mock_makedirs, + mock_git_clone, + mock_verify, + mock_input, + ) -> None: + mock_get_repo_identifier.return_value = "github.com/user/repo" + mock_get_repo_dir.return_value = "/tmp/repos/user/repo" + mock_exists.return_value = False + + # Clone succeeds + mock_git_clone.return_value = None + + # Verification fails, and user answers "n" to proceed anyway + mock_verify.return_value = (False, ["bad signature"], "hash", "key") + + clone_repos( + self.selected, + self.base_dir, + self.all_repos, + preview=False, + no_verification=False, + clone_mode="https", + ) + + mock_git_clone.assert_called_once() + mock_verify.assert_called_once() + mock_input.assert_called_once() if __name__ == "__main__":