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 

11from .runner import CommandResult, run_argv 

12 

13 

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

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

16 

17 

18def worktree_add( 

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

20) -> CommandResult: 

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

22 

23 

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

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

26 

27 

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

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

30 

31 

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

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

34 return result.output.strip() if result.ok else None 

35 

36 

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

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

39 

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

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

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

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

44 """ 

45 return run_argv( 

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

47 "refs/heads", "refs/remotes"], 

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

49 ) 

50 

51 

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

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

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

55 output = result.output.strip() 

56 return output if result.ok and output else None 

57 

58 

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

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

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

62 output = result.output.strip() 

63 return output if result.ok and output else None 

64 

65 

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

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

68 result = run_argv( 

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

70 ) 

71 if not result.ok: 

72 return None 

73 output = result.output.strip() 

74 if not output.isdigit(): 

75 return None 

76 return int(output) 

77 

78 

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

80 """Files changed between ``base`` and ``head`` (``base...head``); ``[]`` on error.""" 

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

82 if not result.ok: 

83 return [] 

84 return [line for line in result.output.splitlines() if line.strip()] 

85 

86 

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

88 """The unified diff between ``base`` and ``head`` (``base...head``); ``""`` on error.""" 

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

90 return result.output if result.ok else "" 

91 

92 

93def _kw(_run): 

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

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