Option C Tools
Loading...
Searching...
No Matches
git.py
Go to the documentation of this file.
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# oct/core/git.py
4
5"""
6Purpose
7-------
8Provide the foundation layer for OCT's Git integration — a pure
9subprocess wrapper and a set of read-only state queries. No CLI, no
10writes, no hook installation. This module is the sole place where
11``subprocess.run`` is allowed to invoke the ``git`` binary; every
12higher-level ``oct git`` command (Phase 4B–4E) delegates through
13``git_run`` and the helpers below.
14
15Responsibilities
16----------------
17- Translate raw subprocess errors into a small, typed ``GitError``
18 hierarchy so callers never see ``FileNotFoundError`` or
19 ``TimeoutExpired``.
20- Enforce subprocess discipline: ``shell=False``, argv-only, explicit
21 timeout, ``text=True``, ``encoding="utf-8"``, ``errors="replace"``.
22- Redact ``user:token@`` segments from any git output that crosses a
23 diagnostics or exception boundary.
24- Provide pure read-only query functions for repo state (branch, HEAD,
25 changed files, staged files, remotes, last commit) each accepting an
26 explicit ``repo_root``.
27- Provide branch mutation primitives (create, switch, create-and-switch)
28 used by ``oct git branch`` and ``oct git commit`` auto-branching.
29- Provide staging primitives (stage, unstage, interactive patch staging)
30 used by ``oct git add``.
31- Filter path-escape attempts in ``git_changed_files`` / ``git_staged_files``
32 by dropping entries whose resolved target lies outside ``repo_root``
33 (OI-405 defense-in-depth).
34
35Diagnostics
36-----------
37Domain: GIT
38Levels:
39 L1 — errors: missing git binary, command failure, path escape
40 L2 — lifecycle: git_run entry/exit with redacted argv + returncode
41 L3 — command: each public query function's entry with arg summary
42 L4 — deep trace: parsed output, per-call timings
43
44Contracts
45---------
46- This module must not import ``click``.
47- Every subprocess invocation goes through :func:`git_run`.
48- ``is_git_repo`` is the only function that swallows all exceptions and
49 returns ``False``; every other public function raises from the
50 ``GitError`` hierarchy on failure.
51- Credential redaction is applied inside ``git_run`` — callers that
52 bubble exception state or log output never see raw credentials.
53- Read-only query functions never write to the repo.
54- Branch mutation functions (``create_branch``, ``switch_branch``,
55 ``create_and_switch_branch``) and staging functions (``stage_paths``,
56 ``unstage_paths``, ``stage_paths_interactive``) are the only write
57 operations and are clearly documented as such.
58- ``stage_paths_interactive`` is the only function that bypasses
59 :func:`git_run` because patch-mode staging requires an attached TTY
60 for the user to answer hunk prompts. It still uses ``shell=False``
61 and validates argv as a list.
62
63Dependencies
64------------
65- oc_diagnostics (_dbg structured logger; optional at dev time)
66"""
67
68import re
69import subprocess
70from pathlib import Path
71from typing import Iterable, Optional
72
73try:
74 from oc_diagnostics import _dbg as _real_dbg
75
76 def _dbg(*args, **kwargs) -> None:
77 _real_dbg(*args, **kwargs)
78except ImportError: # pragma: no cover - diagnostics package is optional at dev time
79 def _dbg(*args, **kwargs) -> None:
80 """No-op fallback when ``oc_diagnostics`` is not importable.
81
82 ``oct.core.git`` is used from unit tests and from the linter,
83 both of which may run in environments where the sibling
84 ``oc_diagnostics`` package is not on the import path (e.g. when
85 pytest is invoked from inside ``oct/`` rather than the project
86 root). The module must remain usable in that case — the tests
87 below exercise behaviour, not logging.
88 """
89 return None
90
91from oct.core.project_root import is_within_project_root
92
93
94# ---------------------------------------------------------------------
95# Module constants
96# ---------------------------------------------------------------------
97
98#: Default timeout for read-only git queries, in seconds. Intentionally
99#: generous so that cold-cache operations on large repositories do not
100#: spuriously raise :class:`GitTimeoutError`.
101DEFAULT_GIT_TIMEOUT: int = 30
102
103#: Regex matching the credential segment of an http(s) URL. Captures the
104#: scheme in group ``scheme`` so we can substitute the credentials with
105#: ``***@`` while preserving the rest of the URL. Intentionally narrow
106#: (no backslashes, no spaces, no slashes, no ``@``) so we only match a
107#: ``user[:pass]@`` prefix and never a trailing ``@`` inside a path.
108_CRED_URL_RE: re.Pattern[str] = re.compile(
109 r"(?P<scheme>https?://)[^/@\s]+@",
110)
111
112
113# ---------------------------------------------------------------------
114# Exception hierarchy (G-A1)
115# ---------------------------------------------------------------------
116
117
118class GitError(Exception):
119 """Base class for every error raised by :mod:`oct.core.git`.
120
121 Callers that want a catch-all should catch ``GitError``; the
122 specific subclasses exist to distinguish "missing binary" from
123 "timeout" from "command failed" so that higher-level CLI commands
124 can render tailored error messages.
125 """
126
127
128class GitNotFoundError(GitError):
129 """Raised when the ``git`` binary is not found on ``PATH``."""
130
131
133 """Raised when a target path is not inside a git working tree."""
134
135
137 """Raised when a git subprocess exceeds its configured timeout."""
138
139
141 """Raised when ``git`` exits with a non-zero status.
142
143 Exposes ``returncode``, ``stdout``, and ``stderr`` attributes. Both
144 ``stdout`` and ``stderr`` are already credential-redacted.
145 """
146
148 self,
149 argv: list[str],
150 returncode: int,
151 stdout: str,
152 stderr: str,
153 ) -> None:
154 # OI-507: redact argv before storing or stringifying so neither the
155 # ``.argv`` attribute nor the exception message can leak credentials.
156 self.argv = _redact_argv(argv)
157 self.returncode = returncode
158 self.stdout = stdout
159 self.stderr = stderr
160 msg = (
161 f"git command failed (exit {returncode}): "
162 f"{' '.join(self.argv)!r}\nstderr: {stderr.strip() or '<empty>'}"
163 )
164 super().__init__(msg)
165
166
167# ---------------------------------------------------------------------
168# Credential redaction (G-A5)
169# ---------------------------------------------------------------------
170
171
172def _redact_git_url(text: str) -> str:
173 """Redact ``user:token@`` segments from http(s) URLs in git output.
174
175 Accepts any string (including multi-line output). Empty strings and
176 ``None`` are returned unchanged. SSH URLs of the form
177 ``git@host:path`` are credential-free (auth is key-based) and are
178 deliberately *not* matched by the regex — they pass through intact.
179
180 The substitution uses ``\\g<scheme>***@`` so the scheme is preserved
181 and the credential segment is replaced with a stable opaque marker.
182 """
183 func = "_redact_git_url"
184 if not text:
185 return text
186 return _CRED_URL_RE.sub(r"\g<scheme>***@", text)
187
188
189def _redact_argv(argv: list[str]) -> list[str]:
190 """Return ``argv`` with credentials redacted in each URL-bearing element.
191
192 OI-507 — ``git_run`` and ``GitCommandError`` both stringify argv for
193 ``_dbg`` and exception messages. Any ``https://user:token@host/...``
194 argument would otherwise land in logs and raised exceptions verbatim.
195 This helper is the single choke point that keeps the redaction
196 invariant (blueprint §11.2) centralised.
197 """
198 return [_redact_git_url(a) for a in argv]
199
200
201# ---------------------------------------------------------------------
202# Subprocess wrapper (G-A2)
203# ---------------------------------------------------------------------
204
205
207 argv: list[str],
208 *,
209 cwd: Path,
210 timeout: int = DEFAULT_GIT_TIMEOUT,
211 check: bool = True,
212) -> subprocess.CompletedProcess[str]:
213 """Run a git command with the project's subprocess discipline.
214
215 Parameters
216 ----------
217 argv
218 Git sub-arguments excluding the leading ``git``. For example,
219 ``["rev-parse", "--show-toplevel"]``. Must be a list; joined
220 strings are rejected to prevent shell injection.
221 cwd
222 Working directory for the git invocation. Must exist.
223 timeout
224 Maximum seconds to wait before the subprocess is killed and
225 :class:`GitTimeoutError` is raised. Defaults to
226 :data:`DEFAULT_GIT_TIMEOUT`.
227 check
228 When ``True`` (default), a non-zero exit code raises
229 :class:`GitCommandError`. When ``False``, the completed process
230 is returned with its ``returncode`` intact so callers can
231 inspect it (used by :func:`is_git_repo` to avoid exception
232 control flow).
233
234 Returns
235 -------
236 ``subprocess.CompletedProcess[str]``
237 A completed process whose ``stdout`` and ``stderr`` have already
238 been passed through :func:`_redact_git_url`.
239
240 Raises
241 ------
242 GitNotFoundError
243 If the ``git`` binary is not on ``PATH``.
244 GitTimeoutError
245 If the subprocess exceeds ``timeout`` seconds.
246 GitCommandError
247 If ``check`` is True and the command exits non-zero.
248 """
249 func = "git_run"
250 if not isinstance(argv, list):
251 raise GitError(f"argv must be a list, got {type(argv).__name__}")
252 full_argv = ["git", *argv]
253 # OI-507: redact argv before logging so token-bearing URLs don't land in _dbg.
254 _dbg("GIT", func, f"argv={_redact_argv(full_argv)!r} cwd={cwd}", 2)
255 try:
256 proc = subprocess.run(
257 full_argv,
258 cwd=str(cwd),
259 shell=False,
260 capture_output=True,
261 text=True,
262 encoding="utf-8",
263 errors="replace",
264 timeout=timeout,
265 )
266 except FileNotFoundError as exc:
267 _dbg("GIT", func, f"git binary not found: {exc}", 1, override=True)
268 raise GitNotFoundError(
269 "The 'git' executable was not found on PATH. "
270 "Install git or adjust PATH."
271 ) from exc
272 except subprocess.TimeoutExpired as exc:
273 # OI-507: both the _dbg line and the raised exception message are
274 # credential-safe via _redact_argv.
275 redacted_argv = _redact_argv(full_argv)
276 _dbg(
277 "GIT", func,
278 f"git timed out after {timeout}s: {' '.join(redacted_argv)!r}",
279 1, override=True,
280 )
281 raise GitTimeoutError(
282 f"git command timed out after {timeout}s: {' '.join(redacted_argv)!r}"
283 ) from exc
284
285 redacted_out = _redact_git_url(proc.stdout or "")
286 redacted_err = _redact_git_url(proc.stderr or "")
287 # Replace the captured streams with their redacted form so any
288 # caller that logs or re-raises cannot accidentally leak credentials.
289 proc.stdout = redacted_out
290 proc.stderr = redacted_err
291
292 _dbg(
293 "GIT", func,
294 f"returncode={proc.returncode} stdout_len={len(redacted_out)} "
295 f"stderr_len={len(redacted_err)}",
296 4,
297 )
298
299 if check and proc.returncode != 0:
300 _dbg(
301 "GIT", func,
302 f"non-zero exit: {proc.returncode} stderr={redacted_err.strip()!r}",
303 1, override=True,
304 )
305 raise GitCommandError(
306 full_argv,
307 returncode=proc.returncode,
308 stdout=redacted_out,
309 stderr=redacted_err,
310 )
311 return proc
312
313
314# ---------------------------------------------------------------------
315# State query functions (G-A3)
316# ---------------------------------------------------------------------
317
318
319def is_git_repo(path: Path) -> bool:
320 """Return True if ``path`` is inside a git working tree.
321
322 This is the only public function in the module that swallows every
323 exception — ``GitNotFoundError``, ``GitCommandError``, ``OSError``,
324 and anything else all collapse to ``False``. Use it as a cheap
325 predicate before calling any other query function.
326 """
327 func = "is_git_repo"
328 _dbg("GIT", func, f"path={path}", 3)
329 try:
330 proc = git_run(
331 ["rev-parse", "--is-inside-work-tree"],
332 cwd=Path(path),
333 check=False,
334 )
335 except GitError:
336 return False
337 except OSError:
338 return False
339 result = proc.returncode == 0 and proc.stdout.strip() == "true"
340 _dbg("GIT", func, f"result={result}", 4)
341 return result
342
343
344def get_repo_root(path: Path) -> Path:
345 """Return the absolute path of the top-level git repo containing ``path``.
346
347 Raises :class:`GitNotARepoError` if ``path`` is not inside a repo.
348 """
349 func = "get_repo_root"
350 _dbg("GIT", func, f"path={path}", 3)
351 proc = git_run(
352 ["rev-parse", "--show-toplevel"],
353 cwd=Path(path),
354 check=False,
355 )
356 if proc.returncode != 0:
357 _dbg("GIT", func, f"not a repo: {path}", 1, override=True)
358 raise GitNotARepoError(f"{path} is not inside a git repository.")
359 root = Path(proc.stdout.strip()).resolve()
360 _dbg("GIT", func, f"root={root}", 4)
361 return root
362
363
364def current_branch(repo_root: Path) -> str:
365 """Return the current branch name, or ``"HEAD"`` if detached.
366
367 Callers that need to distinguish "on a branch" from "detached"
368 should use :func:`is_detached_head` — this function returns the
369 literal string ``"HEAD"`` in the detached case, mirroring git's own
370 ``--abbrev-ref`` output.
371 """
372 func = "current_branch"
373 _dbg("GIT", func, f"repo_root={repo_root}", 3)
374 proc = git_run(
375 ["rev-parse", "--abbrev-ref", "HEAD"],
376 cwd=Path(repo_root),
377 )
378 branch = proc.stdout.strip()
379 _dbg("GIT", func, f"branch={branch!r}", 4)
380 return branch
381
382
383def is_detached_head(repo_root: Path) -> bool:
384 """Return True if HEAD is detached (not pointing at any branch)."""
385 func = "is_detached_head"
386 _dbg("GIT", func, f"repo_root={repo_root}", 3)
387 return current_branch(repo_root) == "HEAD"
388
389
390def has_uncommitted_changes(repo_root: Path) -> bool:
391 """Return True if the working tree has staged or unstaged changes.
392
393 Untracked files are ignored — use :func:`has_untracked_files` for
394 that. This function checks only tracked files via
395 ``git status --porcelain -- uno``.
396 """
397 func = "has_uncommitted_changes"
398 _dbg("GIT", func, f"repo_root={repo_root}", 3)
399 proc = git_run(
400 ["status", "--porcelain", "--untracked-files=no"],
401 cwd=Path(repo_root),
402 )
403 result = bool(proc.stdout.strip())
404 _dbg("GIT", func, f"result={result}", 4)
405 return result
406
407
408def has_untracked_files(repo_root: Path) -> bool:
409 """Return True if there are any untracked files not ignored by .gitignore."""
410 func = "has_untracked_files"
411 _dbg("GIT", func, f"repo_root={repo_root}", 3)
412 proc = git_run(
413 ["ls-files", "--others", "--exclude-standard"],
414 cwd=Path(repo_root),
415 )
416 result = bool(proc.stdout.strip())
417 _dbg("GIT", func, f"result={result}", 4)
418 return result
419
420
422 entries: Iterable[str],
423 repo_root: Path,
424 extensions: Optional[set[str]],
425) -> list[Path]:
426 """Resolve each entry against ``repo_root`` and drop path escapes.
427
428 Internal helper used by :func:`git_changed_files` and
429 :func:`git_staged_files`. Every entry is:
430
431 1. Joined to ``repo_root`` and ``.resolve()``-ed.
432 2. Checked via :func:`is_within_project_root` — any path that
433 escapes the repo (e.g. via a symlink target) is silently dropped
434 with an L1 ``_dbg`` log.
435 3. If ``extensions`` is given, filtered to those suffixes.
436
437 Deduplicates while preserving first-seen order.
438 """
439 func = "_resolve_and_filter"
440 repo_root_resolved = Path(repo_root).resolve()
441 result: list[Path] = []
442 seen: set[Path] = set()
443 for rel in entries:
444 rel = rel.strip()
445 if not rel:
446 continue
447 candidate = (repo_root_resolved / rel).resolve()
448 if not is_within_project_root(candidate, repo_root_resolved):
449 _dbg(
450 "GIT", func,
451 f"dropping path-escape candidate: {rel!r} -> {candidate}",
452 1, override=True,
453 )
454 continue
455 if extensions is not None and candidate.suffix not in extensions:
456 continue
457 if candidate in seen:
458 continue
459 seen.add(candidate)
460 result.append(candidate)
461 return result
462
463
465 repo_root: Path,
466 extensions: Optional[set[str]] = None,
467) -> list[Path]:
468 """Return absolute paths of files changed since the last commit.
469
470 The result is the union of:
471
472 - Tracked files with staged changes (``git diff --cached --name-only``).
473 - Tracked files with unstaged changes (``git diff --name-only``).
474 - Untracked files honouring .gitignore (``git ls-files --others --exclude-standard``).
475
476 If ``extensions`` is supplied (e.g. ``{".py"}``), only files with
477 those suffixes are returned. Paths whose resolved target lies
478 outside ``repo_root`` are silently dropped (defense-in-depth against
479 symlink-escape attacks; each drop is logged at L1).
480 """
481 func = "git_changed_files"
482 _dbg("GIT", func, f"repo_root={repo_root} extensions={extensions}", 3)
483 entries: list[str] = []
484 for argv in (
485 ["diff", "--cached", "--name-only", "--diff-filter=ACMR"],
486 ["diff", "--name-only", "--diff-filter=ACMR"],
487 ["ls-files", "--others", "--exclude-standard"],
488 ):
489 proc = git_run(argv, cwd=Path(repo_root))
490 entries.extend(proc.stdout.splitlines())
491 result = _resolve_and_filter(entries, Path(repo_root), extensions)
492 _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
493 return result
494
495
497 repo_root: Path,
498 extensions: Optional[set[str]] = None,
499) -> list[Path]:
500 """Return absolute paths of files staged for the next commit.
501
502 Same contract as :func:`git_changed_files` but scoped to the index
503 only — untracked and unstaged-but-modified files are excluded.
504 """
505 func = "git_staged_files"
506 _dbg("GIT", func, f"repo_root={repo_root} extensions={extensions}", 3)
507 proc = git_run(
508 ["diff", "--cached", "--name-only", "--diff-filter=ACMR"],
509 cwd=Path(repo_root),
510 )
511 entries = proc.stdout.splitlines()
512 result = _resolve_and_filter(entries, Path(repo_root), extensions)
513 _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
514 return result
515
516
517def last_commit_sha(repo_root: Path, short: bool = False) -> str:
518 """Return the SHA of HEAD, or the empty string on a fresh repo.
519
520 A fresh repo is one where ``git rev-parse HEAD`` fails because no
521 commits exist yet. Any other failure still propagates as
522 :class:`GitCommandError`.
523 """
524 func = "last_commit_sha"
525 _dbg("GIT", func, f"repo_root={repo_root} short={short}", 3)
526 argv = ["rev-parse"]
527 if short:
528 argv.append("--short")
529 argv.append("HEAD")
530 proc = git_run(argv, cwd=Path(repo_root), check=False)
531 if proc.returncode != 0:
532 # Distinguish "no commits yet" from real failures.
533 if "unknown revision" in proc.stderr or "bad revision" in proc.stderr:
534 _dbg("GIT", func, "no commits in repo", 4)
535 return ""
536 raise GitCommandError(
537 ["git", *argv],
538 returncode=proc.returncode,
539 stdout=proc.stdout,
540 stderr=proc.stderr,
541 )
542 sha = proc.stdout.strip()
543 _dbg("GIT", func, f"sha={sha}", 4)
544 return sha
545
546
547def last_commit_message(repo_root: Path) -> str:
548 """Return the subject + body of HEAD, or the empty string on fresh repos.
549
550 Uses ``git log -1 --pretty=%B`` so both the subject and the body are
551 returned together (trailing newline stripped).
552 """
553 func = "last_commit_message"
554 _dbg("GIT", func, f"repo_root={repo_root}", 3)
555 proc = git_run(
556 ["log", "-1", "--pretty=%B"],
557 cwd=Path(repo_root),
558 check=False,
559 )
560 if proc.returncode != 0:
561 if "does not have any commits" in proc.stderr or \
562 "bad default revision" in proc.stderr:
563 _dbg("GIT", func, "no commits in repo", 4)
564 return ""
565 raise GitCommandError(
566 ["git", "log", "-1", "--pretty=%B"],
567 returncode=proc.returncode,
568 stdout=proc.stdout,
569 stderr=proc.stderr,
570 )
571 message = proc.stdout.rstrip("\n")
572 _dbg("GIT", func, f"message_len={len(message)}", 4)
573 return message
574
575
576def remote_urls(repo_root: Path) -> dict[str, str]:
577 """Return a ``{remote_name: url}`` map, with credentials redacted.
578
579 Uses ``git remote -v`` and collapses the fetch + push variants into
580 a single entry per remote (preferring ``fetch`` if both are
581 present). Returns an empty dict if no remotes are configured.
582 """
583 func = "remote_urls"
584 _dbg("GIT", func, f"repo_root={repo_root}", 3)
585 proc = git_run(["remote", "-v"], cwd=Path(repo_root))
586 remotes: dict[str, str] = {}
587 for line in proc.stdout.splitlines():
588 # Format: "<name>\t<url> (fetch|push)"
589 line = line.strip()
590 if not line:
591 continue
592 parts = line.split()
593 if len(parts) < 2:
594 continue
595 name, url = parts[0], parts[1]
596 # Prefer the first-seen URL per remote (git outputs fetch then push).
597 if name not in remotes:
598 remotes[name] = _redact_git_url(url)
599 _dbg("GIT", func, f"remotes={list(remotes.keys())}", 4)
600 return remotes
601
602
603def list_tags(repo_root: Path) -> list[str]:
604 """Return tag names sorted by commit date (newest first).
605
606 Uses ``git tag --sort=-creatordate``. Returns an empty list if no
607 tags exist or if the repo has no commits yet.
608 """
609 func = "list_tags"
610 _dbg("GIT", func, f"repo_root={repo_root}", 3)
611 proc = git_run(
612 ["tag", "--sort=-creatordate"],
613 cwd=Path(repo_root),
614 check=False,
615 )
616 if proc.returncode != 0:
617 _dbg("GIT", func, "no tags or git error", 4)
618 return []
619 tags = [t.strip() for t in proc.stdout.splitlines() if t.strip()]
620 _dbg("GIT", func, f"returning {len(tags)} tag(s)", 4)
621 return tags
622
623
625 repo_root: Path,
626 since_ref: str | None = None,
627 format_str: str = "%H %s",
628) -> list[str]:
629 """Return git log lines from HEAD back to *since_ref* (exclusive).
630
631 Each line is formatted according to *format_str* (passed to
632 ``git log --pretty=<format_str>``). If *since_ref* is ``None``,
633 the entire history is returned. Returns an empty list on fresh
634 repos with no commits.
635 """
636 func = "get_commit_log"
637 _dbg("GIT", func, f"repo_root={repo_root} since_ref={since_ref!r}", 3)
638 argv = ["log", f"--pretty={format_str}"]
639 if since_ref is not None:
640 argv.append(f"{since_ref}..HEAD")
641 proc = git_run(argv, cwd=Path(repo_root), check=False)
642 if proc.returncode != 0:
643 # Fresh repo or invalid ref — return empty.
644 _dbg("GIT", func, "no log output", 4)
645 return []
646 lines = [ln for ln in proc.stdout.splitlines() if ln.strip()]
647 _dbg("GIT", func, f"returning {len(lines)} line(s)", 4)
648 return lines
649
650
652 repo_root: Path,
653 ref: str,
654 extensions: Optional[set[str]] = None,
655) -> list[Path]:
656 """Return absolute paths of files changed between *ref* and HEAD.
657
658 Uses ``git diff --name-only <ref>`` and applies the same
659 resolve-and-filter pipeline as :func:`git_changed_files` to drop
660 path-escape attempts and optionally filter by extension.
661 """
662 func = "git_diff_name_only"
663 _dbg("GIT", func, f"repo_root={repo_root} ref={ref!r}", 3)
664 proc = git_run(
665 ["diff", "--name-only", ref],
666 cwd=Path(repo_root),
667 check=False,
668 )
669 if proc.returncode != 0:
670 _dbg("GIT", func, "diff failed — returning empty", 4)
671 return []
672 entries = proc.stdout.splitlines()
673 result = _resolve_and_filter(entries, Path(repo_root), extensions)
674 _dbg("GIT", func, f"returning {len(result)} path(s)", 4)
675 return result
676
677
678def ahead_behind(repo_root: Path) -> tuple[int, int]:
679 """Return ``(ahead, behind)`` counts relative to the upstream branch.
680
681 Uses ``git rev-list --left-right --count HEAD...@{upstream}``.
682 Returns ``(0, 0)`` if no upstream is configured or on a fresh repo.
683 """
684 func = "ahead_behind"
685 _dbg("GIT", func, f"repo_root={repo_root}", 3)
686 proc = git_run(
687 ["rev-list", "--left-right", "--count", "HEAD...@{upstream}"],
688 cwd=Path(repo_root),
689 check=False,
690 )
691 if proc.returncode != 0:
692 _dbg("GIT", func, "no upstream or error — returning (0, 0)", 4)
693 return 0, 0
694 parts = proc.stdout.strip().split()
695 if len(parts) != 2:
696 return 0, 0
697 try:
698 ahead = int(parts[0])
699 behind = int(parts[1])
700 except ValueError:
701 return 0, 0
702 _dbg("GIT", func, f"ahead={ahead} behind={behind}", 4)
703 return ahead, behind
704
705
706# ---------------------------------------------------------------------
707# Branch mutation functions (G-F1)
708# ---------------------------------------------------------------------
709
710
711def branch_exists(repo_root: Path, name: str) -> bool:
712 """Return True if a local branch *name* exists.
713
714 Uses ``git rev-parse --verify refs/heads/<name>`` with
715 ``check=False`` so it never raises on a missing branch.
716 """
717 func = "branch_exists"
718 _dbg("GIT", func, f"repo_root={repo_root} name={name!r}", 3)
719 proc = git_run(
720 ["rev-parse", "--verify", f"refs/heads/{name}"],
721 cwd=Path(repo_root),
722 check=False,
723 )
724 result = proc.returncode == 0
725 _dbg("GIT", func, f"result={result}", 4)
726 return result
727
728
730 repo_root: Path,
731 name: str,
732 start_point: str | None = None,
733) -> None:
734 """Create a new branch *name* pointing at *start_point* (default HEAD).
735
736 Does **not** switch to the new branch — use :func:`switch_branch` or
737 :func:`create_and_switch_branch` for that.
738
739 Raises :class:`GitCommandError` if the branch already exists or the
740 start-point is invalid.
741 """
742 func = "create_branch"
743 _dbg("GIT", func, f"name={name!r} start_point={start_point!r}", 3)
744 argv = ["branch", name]
745 if start_point:
746 argv.append(start_point)
747 git_run(argv, cwd=Path(repo_root))
748 _dbg("GIT", func, f"created branch {name!r}", 4)
749
750
751def switch_branch(repo_root: Path, name: str) -> None:
752 """Switch the working tree to an existing branch *name*.
753
754 Uses ``git checkout`` for maximum git-version compatibility.
755 Raises :class:`GitCommandError` if the branch does not exist or
756 the checkout fails (e.g. uncommitted changes would be overwritten).
757 """
758 func = "switch_branch"
759 _dbg("GIT", func, f"name={name!r}", 3)
760 git_run(["checkout", name], cwd=Path(repo_root))
761 _dbg("GIT", func, f"switched to {name!r}", 4)
762
763
765 repo_root: Path,
766 name: str,
767 start_point: str | None = None,
768) -> None:
769 """Atomically create and switch to a new branch *name*.
770
771 Equivalent to ``git checkout -b <name> [start_point]``. The index
772 and working tree are preserved — staged changes carry over to the
773 new branch.
774
775 Raises :class:`GitCommandError` if the branch already exists.
776 """
777 func = "create_and_switch_branch"
778 _dbg("GIT", func, f"name={name!r} start_point={start_point!r}", 3)
779 argv = ["checkout", "-b", name]
780 if start_point:
781 argv.append(start_point)
782 git_run(argv, cwd=Path(repo_root))
783 _dbg("GIT", func, f"created and switched to {name!r}", 4)
784
785
786def list_branches(repo_root: Path, *, remote: bool = False) -> list[str]:
787 """Return a list of local (or remote) branch names.
788
789 Uses ``git branch --format=%(refname:short)`` for clean output
790 without the ``*`` marker or whitespace prefix.
791 """
792 func = "list_branches"
793 _dbg("GIT", func, f"repo_root={repo_root} remote={remote}", 3)
794 argv = ["branch", "--format=%(refname:short)"]
795 if remote:
796 argv.insert(1, "-r")
797 proc = git_run(argv, cwd=Path(repo_root))
798 branches = [b.strip() for b in proc.stdout.splitlines() if b.strip()]
799 _dbg("GIT", func, f"returning {len(branches)} branch(es)", 4)
800 return branches
801
802
804 repo_root: Path,
805 paths: list[Path] | list[str],
806 *,
807 update: bool = False,
808 all_: bool = False,
809) -> None:
810 """Stage *paths* for the next commit (whole-file staging).
811
812 Equivalent to ``git add [<paths>...]`` with optional ``-u`` (update
813 tracked only) or ``-A`` (all changes including untracked).
814
815 Parameters
816 ----------
817 repo_root
818 Git repository root.
819 paths
820 Paths to stage. May be empty when ``update`` or ``all_`` is set.
821 update
822 Pass ``-u`` so only already-tracked files are staged.
823 all_
824 Pass ``-A`` so untracked files are also staged.
825
826 Raises
827 ------
828 GitCommandError
829 If git rejects the operation (e.g. path outside repo).
830 """
831 func = "stage_paths"
832 _dbg(
833 "GIT", func,
834 f"repo_root={repo_root} paths={len(paths)} update={update} all={all_}",
835 3,
836 )
837 argv: list[str] = ["add"]
838 if update:
839 argv.append("-u")
840 if all_:
841 argv.append("-A")
842 if paths:
843 argv.append("--")
844 argv.extend(str(p) for p in paths)
845 elif not (update or all_):
846 # Nothing to do; refuse silently rather than running ``git add``
847 # with no arguments which is a no-op in modern git but warns.
848 _dbg("GIT", func, "no paths and no -u/-A; nothing to stage", 4)
849 return
850 git_run(argv, cwd=Path(repo_root))
851 _dbg("GIT", func, "stage complete", 4)
852
853
855 repo_root: Path,
856 paths: list[Path] | list[str],
857) -> None:
858 """Remove *paths* from the index (unstage), leaving working tree intact.
859
860 Equivalent to ``git reset HEAD -- <paths>``. Provided for symmetry
861 with :func:`stage_paths`; not currently exposed via CLI but used in
862 tests and reserved for a future ``oct git unstage`` command.
863
864 Raises
865 ------
866 GitCommandError
867 If git rejects the operation.
868 """
869 func = "unstage_paths"
870 _dbg("GIT", func, f"repo_root={repo_root} paths={len(paths)}", 3)
871 if not paths:
872 _dbg("GIT", func, "no paths; nothing to unstage", 4)
873 return
874 argv: list[str] = ["reset", "HEAD", "--"]
875 argv.extend(str(p) for p in paths)
876 git_run(argv, cwd=Path(repo_root))
877 _dbg("GIT", func, "unstage complete", 4)
878
879
881 repo_root: Path,
882 paths: list[Path] | list[str],
883 *,
884 staged: bool = False,
885) -> None:
886 """Restore *paths* from the index (working tree) or HEAD (staged).
887
888 L3 / Phase 4G-followup. Equivalent to:
889
890 - ``staged=False``: ``git restore <paths>`` — drop unstaged
891 working-tree edits, restoring from the index.
892 - ``staged=True``: ``git restore --staged <paths>`` — unstage
893 *paths* without touching the working tree.
894
895 Raises
896 ------
897 GitCommandError
898 If git rejects the operation.
899 """
900 func = "restore_paths"
901 _dbg(
902 "GIT", func,
903 f"repo_root={repo_root} paths={len(paths)} staged={staged}",
904 3,
905 )
906 if not paths:
907 _dbg("GIT", func, "no paths; nothing to restore", 4)
908 return
909 argv: list[str] = ["restore"]
910 if staged:
911 argv.append("--staged")
912 argv.append("--")
913 argv.extend(str(p) for p in paths)
914 git_run(argv, cwd=Path(repo_root))
915 _dbg("GIT", func, "restore complete", 4)
916
917
919 repo_root: Path,
920 paths: list[Path] | list[str],
921) -> int:
922 """Run ``git add -p`` so the user can choose hunks interactively.
923
924 This is the only public function in this module that bypasses
925 :func:`git_run`. Patch-mode staging requires an attached TTY so the
926 user can answer hunk prompts (``y/n/s/...``); ``git_run`` always
927 captures stdout/stderr which would break that interaction. The
928 bypass still uses ``shell=False`` and validates argv.
929
930 Parameters
931 ----------
932 repo_root
933 Git repository root.
934 paths
935 Optional list of paths to scope the patch session to. When
936 empty, ``git`` walks every modified tracked file.
937
938 Returns
939 -------
940 int
941 Process exit code from ``git add -p``. ``0`` on success.
942
943 Raises
944 ------
945 GitNotFoundError
946 If the ``git`` executable is not on PATH.
947 """
948 func = "stage_paths_interactive"
949 _dbg("GIT", func, f"repo_root={repo_root} paths={len(paths)}", 3)
950 argv: list[str] = ["git", "add", "-p"]
951 if paths:
952 argv.append("--")
953 argv.extend(str(p) for p in paths)
954 if not isinstance(argv, list):
955 raise GitError(
956 f"argv must be a list, got {type(argv).__name__}",
957 )
958 try:
959 proc = subprocess.run(
960 argv,
961 cwd=str(repo_root),
962 shell=False,
963 check=False,
964 )
965 except FileNotFoundError as exc:
966 _dbg("GIT", func, f"git binary not found: {exc}", 1, override=True)
967 raise GitNotFoundError(
968 "The 'git' executable was not found on PATH. "
969 "Install git or adjust PATH."
970 ) from exc
971 _dbg("GIT", func, f"interactive stage exit={proc.returncode}", 4)
972 return proc.returncode
None __init__(self, list[str] argv, int returncode, str stdout, str stderr)
Definition git.py:153
tuple[int, int] ahead_behind(Path repo_root)
Definition git.py:678
None stage_paths(Path repo_root, list[Path]|list[str] paths, *, bool update=False, bool all_=False)
Definition git.py:809
bool is_detached_head(Path repo_root)
Definition git.py:383
None restore_paths(Path repo_root, list[Path]|list[str] paths, *, bool staged=False)
Definition git.py:885
bool branch_exists(Path repo_root, str name)
Definition git.py:711
list[str] _redact_argv(list[str] argv)
Definition git.py:189
None unstage_paths(Path repo_root, list[Path]|list[str] paths)
Definition git.py:857
str _redact_git_url(str text)
Definition git.py:172
list[str] list_branches(Path repo_root, *, bool remote=False)
Definition git.py:786
bool is_git_repo(Path path)
Definition git.py:319
list[Path] git_changed_files(Path repo_root, Optional[set[str]] extensions=None)
Definition git.py:467
bool has_uncommitted_changes(Path repo_root)
Definition git.py:390
dict[str, str] remote_urls(Path repo_root)
Definition git.py:576
str current_branch(Path repo_root)
Definition git.py:364
Path get_repo_root(Path path)
Definition git.py:344
None _dbg(*args, **kwargs)
Definition git.py:76
subprocess.CompletedProcess[str] git_run(list[str] argv, *, Path cwd, int timeout=DEFAULT_GIT_TIMEOUT, bool check=True)
Definition git.py:212
list[Path] git_staged_files(Path repo_root, Optional[set[str]] extensions=None)
Definition git.py:499
list[Path] git_diff_name_only(Path repo_root, str ref, Optional[set[str]] extensions=None)
Definition git.py:655
list[str] list_tags(Path repo_root)
Definition git.py:603
None switch_branch(Path repo_root, str name)
Definition git.py:751
int stage_paths_interactive(Path repo_root, list[Path]|list[str] paths)
Definition git.py:921
None create_branch(Path repo_root, str name, str|None start_point=None)
Definition git.py:733
str last_commit_sha(Path repo_root, bool short=False)
Definition git.py:517
list[str] get_commit_log(Path repo_root, str|None since_ref=None, str format_str="%H %s")
Definition git.py:628
str last_commit_message(Path repo_root)
Definition git.py:547
None create_and_switch_branch(Path repo_root, str name, str|None start_point=None)
Definition git.py:768
bool has_untracked_files(Path repo_root)
Definition git.py:408
list[Path] _resolve_and_filter(Iterable[str] entries, Path repo_root, Optional[set[str]] extensions)
Definition git.py:425