refactor(pull): switch repository pull to core.git commands
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

- Replace raw subprocess git pull with core.git.commands.pull_args
- Remove shell-based command execution
- Add GitPullArgsError wrapper for consistent error handling
- Align unit tests to mock pull_args instead of subprocess.run
- Preserve verification and prompt logic

https://chatgpt.com/share/69414dc9-5b30-800f-88b2-bd27a873580b
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-16 13:17:04 +01:00
parent 019aa4b0d9
commit 2f89de1ff5
4 changed files with 121 additions and 80 deletions

View File

@@ -1,25 +1,30 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import os
import subprocess
import sys
from typing import List, Dict, Any
from pkgmgr.core.git.commands import pull_args, GitPullArgsError
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 pull_with_verification(
selected_repos,
repositories_base_dir,
all_repos,
extra_args,
no_verification,
selected_repos: List[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
extra_args: List[str],
no_verification: bool,
preview: bool,
) -> None:
"""
Execute `git pull` for each repository with verification.
- If verification fails and verification is enabled, prompt user to continue.
- Uses core.git.commands.pull_args() (no raw subprocess usage).
"""
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
@@ -37,12 +42,7 @@ def pull_with_verification(
no_verification=no_verification,
)
if (
not preview
and not no_verification
and verified_info
and not verified_ok
):
if not preview and not no_verification and verified_info and not verified_ok:
print(f"Warning: Verification failed for {repo_identifier}:")
for err in errors:
print(f" - {err}")
@@ -50,17 +50,10 @@ def pull_with_verification(
if choice != "y":
continue
args_part = " ".join(extra_args) if extra_args else ""
full_cmd = f"git pull{(' ' + args_part) if args_part else ''}"
if preview:
print(f"[Preview] In '{repo_dir}': {full_cmd}")
else:
print(f"Running in '{repo_dir}': {full_cmd}")
result = subprocess.run(full_cmd, cwd=repo_dir, shell=True, check=False)
if result.returncode != 0:
print(
f"'git pull' for {repo_identifier} failed "
f"with exit code {result.returncode}."
)
sys.exit(result.returncode)
try:
pull_args(extra_args, cwd=repo_dir, preview=preview)
except GitPullArgsError as exc:
# Keep behavior consistent with previous implementation:
# stop on first failure and propagate return code as generic failure.
print(str(exc))
sys.exit(1)

View File

@@ -16,6 +16,7 @@ from .fetch import GitFetchError, fetch
from .init import GitInitError, init
from .merge_no_ff import GitMergeError, merge_no_ff
from .pull import GitPullError, pull
from .pull_args import GitPullArgsError, pull_args # <-- add
from .pull_ff_only import GitPullFfOnlyError, pull_ff_only
from .push import GitPushError, push
from .push_upstream import GitPushUpstreamError, push_upstream
@@ -29,6 +30,7 @@ __all__ = [
"fetch",
"checkout",
"pull",
"pull_args", # <-- add
"pull_ff_only",
"merge_no_ff",
"push",
@@ -50,6 +52,7 @@ __all__ = [
"GitFetchError",
"GitCheckoutError",
"GitPullError",
"GitPullArgsError", # <-- add
"GitPullFfOnlyError",
"GitMergeError",
"GitPushError",

View File

@@ -0,0 +1,35 @@
from __future__ import annotations
from typing import List
from ..errors import GitError, GitCommandError
from ..run import run
class GitPullArgsError(GitCommandError):
"""Raised when `git pull` with arbitrary args fails."""
def pull_args(
args: List[str] | None = None,
*,
cwd: str = ".",
preview: bool = False,
) -> None:
"""
Execute `git pull` with caller-provided arguments.
Examples:
[] -> git pull
["--ff-only"] -> git pull --ff-only
["--rebase"] -> git pull --rebase
["origin", "main"] -> git pull origin main
"""
extra = args or []
try:
run(["pull", *extra], cwd=cwd, preview=preview)
except GitError as exc:
raise GitPullArgsError(
f"Failed to run `git pull` with args={extra!r}.",
cwd=cwd,
) from exc