Coverage for src/keel/git.py: 100%

1"""Thin, fail-soft ``git`` wrappers (argv, no shell). 

2 

3These build the exact git command for each backbone operation and run it via the 

4injectable ``_run`` seam, so the command construction is unit-tested offline; live 

5behaviour is exercised opt-in against a real repo. Each returns a 

6:class:`keel.runner.CommandResult` (or a parsed value), never raising. 

7""" 

8 

9from __future__ import annotations 

10 

11import re 

12 

13from .runner import CommandResult, run_argv 

14 

15 

16def fetch(remote: str, ref: str, *, cwd: str | None = None, _run=None) -> CommandResult: 

17 return run_argv(["git", "fetch", remote, ref, "--quiet"], cwd=cwd, **_kw(_run)) 

18 

19 

20def worktree_add( 

21 base: str, branch: str, path: str, *, cwd: str | None = None, _run=None 

22) -> CommandResult: 

23 return run_argv(["git", "worktree", "add", "-b", branch, path, base], cwd=cwd, **_kw(_run)) 

24 

25 

26def worktree_remove(path: str, *, cwd: str | None = None, _run=None) -> CommandResult: 

27 return run_argv(["git", "worktree", "remove", path, "--force"], cwd=cwd, **_kw(_run)) 

28 

29 

30def worktree_list(*, cwd: str | None = None, _run=None) -> CommandResult: 

31 return run_argv(["git", "worktree", "list", "--porcelain"], cwd=cwd, **_kw(_run)) 

32 

33 

34def current_branch(*, cwd: str | None = None, _run=None) -> str | None: 

35 result = run_argv(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=cwd, **_kw(_run)) 

36 return result.stdout.strip() if result.ok else None 

37 

38 

39def list_branches(*, cwd: str | None = None, _run=None) -> CommandResult: 

40 """List local + remote branch short names (one per line) as a ``CommandResult``. 

41 

42 Returns the raw result (like :func:`worktree_list`) rather than a parsed 

43 fail-soft list, so a caller that needs to *distinguish a git error from an 

44 empty repo* — e.g. dry-run integrity verification, which must fail closed 

45 when it cannot observe — can inspect ``result.ok``. Parsing is the caller's. 

46 """ 

47 return run_argv( 

48 ["git", "for-each-ref", "--format=%(refname:short)", 

49 "refs/heads", "refs/remotes"], 

50 cwd=cwd, **_kw(_run), 

51 ) 

52 

53 

54#: A 40- or 64-hex object name (SHA-1 / SHA-256). git may print a ``warning:`` to 

55#: stderr while still succeeding; reading ``stdout`` avoids the contamination, and 

56#: validating the shape is a second line of defence so a stray token never poses as a SHA. 

57_SHA_RE = re.compile(r"\A[0-9a-f]{40}(?:[0-9a-f]{24})?\Z") 

58 

59 

60def rev_parse(ref: str, *, cwd: str | None = None, _run=None) -> str | None: 

61 """Resolve ``ref`` to a full commit SHA; ``None`` when it cannot be resolved.""" 

62 result = run_argv(["git", "rev-parse", "--verify", "--quiet", ref], cwd=cwd, **_kw(_run)) 

63 output = result.stdout.strip() 

64 return output if result.ok and _SHA_RE.match(output) else None 

65 

66 

67def merge_base(a: str, b: str, *, cwd: str | None = None, _run=None) -> str | None: 

68 """Best common ancestor of ``a`` and ``b``; ``None`` when there is none/on error.""" 

69 result = run_argv(["git", "merge-base", a, b], cwd=cwd, **_kw(_run)) 

70 output = result.stdout.strip() 

71 return output if result.ok and _SHA_RE.match(output) else None 

72 

73 

74def rev_count(base: str, head: str, *, cwd: str | None = None, _run=None) -> int | None: 

75 """Commits in ``base..head`` (how far ``head`` is ahead of ``base``); ``None`` on error.""" 

76 result = run_argv( 

77 ["git", "rev-list", "--count", f"{base}..{head}"], cwd=cwd, **_kw(_run) 

78 ) 

79 if not result.ok: 

80 return None 

81 output = result.stdout.strip() 

82 if not output.isdigit(): 

83 return None 

84 return int(output) 

85 

86 

87def changed_files(base: str, head: str, *, cwd: str | None = None, _run=None) -> list[str] | None: 

88 """Files changed between ``base`` and ``head`` (``base...head``). 

89 

90 ``None`` when the git command failed — deliberately distinct from ``[]`` (the 

91 command ran and there were no changes), so a caller classifying risk or checking 

92 scope can tell "could not read the diff" apart from "the diff is empty" instead of 

93 treating an unreadable diff as a clean, empty one. 

94 """ 

95 result = run_argv(["git", "diff", "--name-only", f"{base}...{head}"], cwd=cwd, **_kw(_run)) 

96 if not result.ok: 

97 return None 

98 return [line for line in result.stdout.splitlines() if line.strip()] 

99 

100 

101def diff(base: str, head: str, *, cwd: str | None = None, _run=None) -> str | None: 

102 """The unified diff between ``base`` and ``head`` (``base...head``). 

103 

104 ``None`` when the git command failed — distinct from ``""`` (the command ran and 

105 the diff is empty), so a review/gate caller can refuse to treat an unreadable diff 

106 as "nothing to review". 

107 """ 

108 result = run_argv(["git", "diff", f"{base}...{head}"], cwd=cwd, **_kw(_run)) 

109 return result.stdout if result.ok else None 

110 

111 

112def _kw(_run): 

113 """Pass ``_run`` through only when provided (so the default subprocess is used otherwise).""" 

114 return {"_run": _run} if _run is not None else {}