Coverage for vibefpmgui/git_status.py: 95%
163 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-01 20:51 +0000
1"""Git: a project's repository's status, and the two versions of a changed
2file to compare, from the git program (``git status --porcelain=v2``, a
3format made for programs; ``git cat-file``). Git must be installed (on the
4PATH); without it the Git panel says so.
6git runs as a process of its own (see `vibefpmgui.console.run`): this
7process only waits for it, without Python's lock, so the window goes on.
9No Qt: `vibefpmgui.git_panel` shows it, `vibefpmgui.window.git` reads the
10status on a pool thread.
11"""
13from __future__ import annotations
15import os
16import shutil
17from dataclasses import dataclass, field
18from pathlib import Path
20from . import console
21from .source_text import decode
23# Where a change is: what `git status` calls "Changes to be committed",
24# "Changes not staged for commit", "Untracked files" and "Unmerged paths".
25STAGED, CHANGED, UNTRACKED, CONFLICT = "staged", "changed", "untracked", "conflict"
26AREAS = (CONFLICT, STAGED, CHANGED, UNTRACKED) # (the order they're listed in)
27AREA_TITLES = {CONFLICT: "Conflicts", STAGED: "Staged Changes", CHANGED: "Changes", UNTRACKED: "Untracked"}
29# What happened to the file, and its letter (as in `git status --short`).
30LETTERS = {"modified": "M", "added": "A", "deleted": "D", "renamed": "R", "typechange": "T",
31 "untracked": "U", "conflict": "!"}
32KIND_WORDS = {"modified": "modified", "added": "added", "deleted": "deleted", "renamed": "renamed",
33 "typechange": "type changed", "untracked": "untracked", "conflict": "conflict"}
35MAX_DIFF_BYTES = 4 * 1024 * 1024 # (a bigger version isn't shown: see versions)
36TIMEOUT = 120 # seconds a git command may take
37BINARY = "\0binary" # versions() marker: the file isn't text
40@dataclass(frozen=True)
41class Change:
42 path: str # relative to the repository's top folder, with "/"
43 area: str # STAGED, CHANGED, UNTRACKED or CONFLICT
44 kind: str # a key of LETTERS
45 old_path: str | None = None # a renamed file's name before
47 @property
48 def letter(self) -> str:
49 return LETTERS[self.kind]
52@dataclass
53class RepoStatus:
54 root: Path # the repository's top folder (its work tree)
55 branch: str | None = None # None: detached HEAD (see head)
56 head: str | None = None # the commit checked out (short), None before the first commit
57 upstream: str | None = None # e.g. "origin/main"
58 ahead: int = 0 # commits here that the upstream doesn't have
59 behind: int = 0 # and the other way around
60 changes: list[Change] = field(default_factory=list)
62 def in_area(self, area: str) -> list[Change]:
63 return [c for c in self.changes if c.area == area]
65 def summary(self) -> str:
66 """ "main → origin/main · 2 ahead, 1 behind" and the like."""
67 if self.branch is None:
68 where = f"detached at {self.head}" if self.head else "detached HEAD"
69 else:
70 where = self.branch + ("" if self.head else " (no commits yet)")
71 if self.upstream:
72 where += f" → {self.upstream}"
73 counts = [f"{n} {word}" for n, word in ((self.ahead, "ahead"), (self.behind, "behind")) if n]
74 where += f" · {', '.join(counts) if counts else 'up to date'}"
75 return where
78class GitError(RuntimeError):
79 """git couldn't do it (its message)."""
82def executable() -> str | None:
83 """The git program, or None if it isn't installed (on the PATH)."""
84 return shutil.which("git")
87def available() -> bool:
88 """Whether git is installed (the Git panel says so otherwise)."""
89 return executable() is not None
92def _git(folder: Path, *args: str, limit: int | None = None) -> tuple[int, bytes | None, str]:
93 """Run ``git -C folder args`` (no optional locks: it never holds the
94 index lock that your own git commands need); (exit code, its output,
95 or None if over ``limit`` bytes, its error message)."""
96 git = executable()
97 if git is None: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 raise GitError("git isn't installed")
99 env = dict(os.environ, GIT_OPTIONAL_LOCKS="0", GIT_TERMINAL_PROMPT="0")
100 try:
101 code, output, message = console.run([git, "-C", str(folder), *args], TIMEOUT, limit, env)
102 except OSError as error:
103 raise GitError(f"couldn't run git: {error}") from None
104 except TimeoutError as error:
105 raise GitError(str(error)) from None
106 return code, output, message
109def _check(folder: Path, *args: str) -> bytes:
110 code, output, message = _git(folder, *args)
111 if code != 0:
112 lines = message.splitlines()
113 raise GitError(lines[-1].removeprefix("fatal: ") if lines else f"git {args[0]} failed ({code})")
114 return output or b""
117def find_repository(path: Path) -> Path | None:
118 """The top folder of the git repository ``path`` is in, or None (not in
119 one, a bare one, or no git)."""
120 if not available() or not Path(path).is_dir():
121 return None
122 code, output, _message = _git(Path(path), "rev-parse", "--show-toplevel")
123 if code != 0 or not output or not output.strip():
124 return None
125 return Path(output.decode("utf-8", "replace").strip()).resolve()
128# Each side of porcelain v2's "XY": what happened (index side: staged; work tree side: not staged).
129KINDS = {"M": "modified", "A": "added", "D": "deleted", "R": "renamed", "C": "added", "T": "typechange"}
132def read_status(path: Path) -> RepoStatus | None:
133 """The status of the repository ``path`` is in (None: not in one). A file
134 can be in two areas: some of its changes staged, some not. Raises
135 GitError if git can't read it (its message)."""
136 root = find_repository(path)
137 if root is None:
138 return None
139 output = _check(root, "status", "--porcelain=v2", "--branch", "-z", "--untracked-files=all")
140 return parse_status(root, output)
143def discard(root: Path, change: Change) -> list[Path]:
144 """Discard a file's changes (the Git panel's Discard Changes…). In
145 Changes: back to its staged version (or the last commit's), what's staged
146 kept. In Staged Changes: back to the last commit's version, staged and
147 not (a file new in it unstaged, a renamed one back under its old name).
148 The files then to remove, which git has no copy of (an untracked file, a
149 new one unstaged): returned, not deleted here (see the window: they go
150 to the Trash). Raises GitError (git's message)."""
151 if change.area == CONFLICT:
152 raise GitError("a file with a conflict: resolve it, or abort the merge, first")
153 if change.area == UNTRACKED:
154 return [root / change.path]
155 if change.area == CHANGED:
156 _check(root, "restore", "--worktree", "--", change.path)
157 return []
158 if change.kind in ("added", "renamed"): # (not in the last commit under this name)
159 _check(root, "rm", "--cached", "--quiet", "--", change.path)
160 if change.kind == "renamed" and change.old_path:
161 _check(root, "restore", "--source=HEAD", "--staged", "--worktree", "--", change.old_path)
162 return [root / change.path]
163 _check(root, "restore", "--source=HEAD", "--staged", "--worktree", "--", change.path)
164 return []
167def parse_status(root: Path, output: bytes) -> RepoStatus:
168 """``git status --porcelain=v2 --branch -z`` read."""
169 status = RepoStatus(root)
170 fields = output.decode("utf-8", "surrogateescape").split("\0")
171 i = 0
172 while i < len(fields):
173 entry = fields[i]
174 i += 1
175 if entry.startswith("# "):
176 key, _, value = entry[2:].partition(" ")
177 if key == "branch.oid":
178 status.head = None if value == "(initial)" else value[:7]
179 elif key == "branch.head":
180 status.branch = None if value == "(detached)" else value
181 elif key == "branch.upstream":
182 status.upstream = value
183 elif key == "branch.ab": 183 ↛ 172line 183 didn't jump to line 172 because the condition on line 183 was always true
184 ahead, _, behind = value.partition(" ")
185 status.ahead, status.behind = abs(int(ahead)), abs(int(behind))
186 elif entry.startswith(("1 ", "2 ")):
187 parts = entry.split(" ", 9 if entry[0] == "2" else 8)
188 file, old = parts[-1], None
189 if entry[0] == "2": # (renamed or copied: the next field is its name before)
190 old = fields[i]
191 i += 1
192 staged, changed = parts[1][0], parts[1][1]
193 if staged != ".":
194 status.changes.append(Change(file, STAGED, KINDS.get(staged, "modified"),
195 old if staged in "RC" else None))
196 if changed != ".":
197 status.changes.append(Change(file, CHANGED, KINDS.get(changed, "modified")))
198 elif entry.startswith("u "):
199 status.changes.append(Change(entry.split(" ", 10)[-1], CONFLICT, "conflict"))
200 elif entry.startswith("? "):
201 status.changes.append(Change(entry[2:], UNTRACKED, "untracked"))
202 status.changes.sort(key=lambda c: c.path)
203 return status
206def _text(data: bytes | None) -> str | None:
207 if data is None:
208 return None
209 if len(data) > MAX_DIFF_BYTES or b"\0" in data[:8192]:
210 return BINARY
211 return decode(data)[0].replace("\r\n", "\n")
214def versions(root: Path, change: Change) -> tuple[str | None, str | None, str, str]:
215 """(old text, new text, old label, new label) of a change: what `git diff`
216 compares. Staged: the last commit and the index; not staged: the index
217 and the file; untracked: nothing and the file; a conflict: the last
218 commit and the file (with its conflict markers). None: there's no such
219 version (added, deleted); BINARY: it isn't text, or it's huge."""
221 def blob(spec: str) -> bytes | str | None:
222 code, output, _message = _git(root, "cat-file", "blob", spec, limit=MAX_DIFF_BYTES)
223 if code != 0: 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true
224 return None # (not in that version)
225 return BINARY if output is None else output
227 def committed() -> bytes | str | None:
228 return blob(f"HEAD:{change.old_path or change.path}")
230 def indexed() -> bytes | str | None:
231 return blob(f":{change.path}")
233 def on_disk() -> bytes | str | None:
234 file = root / change.path
235 try:
236 if not file.is_file():
237 return None
238 return BINARY if file.stat().st_size > MAX_DIFF_BYTES else file.read_bytes()
239 except OSError:
240 return None
242 if change.area == STAGED:
243 old, new, labels = committed(), indexed(), ("last commit", "staged")
244 elif change.area == CHANGED:
245 old, new, labels = indexed(), on_disk(), ("staged", "working tree")
246 elif change.area == UNTRACKED:
247 old, new, labels = None, on_disk(), ("", "working tree")
248 else:
249 old, new, labels = committed(), on_disk(), ("last commit", "working tree")
250 return (*(v if isinstance(v, str) else _text(v) for v in (old, new)), *labels)