Coverage for vibefpmgui/git_log.py: 94%

134 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-11-01 20:51 +0000

1"""The Git panel's *History*: the repository's commits, newest first, with 

2their branches and tags, and the graph of how they join (each commit's 

3lane, the lines to its parents), and the two versions of a file a commit 

4changed. 

5 

6The graph's layout (`graph`): each lane waits for a commit (a parent of one 

7above it). A commit goes in the lane waiting for it (the leftmost, if 

8several are: the others join it there); then that lane waits for its first 

9parent, and its other parents (a merge's) get lanes of their own, the 

10leftmost free one, unless one is already waiting for them. Lanes keep their 

11places (a free one is used again), so a line that passes a row goes 

12straight down; each lane keeps its color from where it started. 

13 

14No Qt: `vibefpmgui.git_log_view` shows it; it's read on a pool thread. 

15""" 

16 

17from __future__ import annotations 

18 

19from dataclasses import dataclass, field 

20from pathlib import Path 

21 

22from .git_blame import FileChange 

23from .git_status import BINARY, MAX_DIFF_BYTES, _check, _git, _text, find_repository 

24 

25MAX_COMMITS = 3000 # read at first, at most (the newest; the rest: the History's Load Older Commits) 

26 

27# Ref kinds, as their badges show them. 

28HEAD, BRANCH, REMOTE, TAG = "head", "branch", "remote", "tag" 

29 

30 

31@dataclass 

32class Ref: 

33 name: str # "main", "origin/main", "v1.0", "HEAD" (detached) 

34 kind: str # HEAD, BRANCH, REMOTE or TAG 

35 current: bool = False # the branch checked out (HEAD -> it) 

36 

37 

38@dataclass 

39class LogCommit: 

40 sha: str 

41 parents: list[str] 

42 author: str 

43 email: str 

44 time: int # its author's date (seconds since 1970) 

45 subject: str 

46 refs: list[Ref] = field(default_factory=list) 

47 

48 

49@dataclass 

50class GraphRow: 

51 """A commit's row of the graph: lanes are numbered from the left, each 

52 line given as (lane, color number).""" 

53 lane: int # the commit's 

54 color: int # its lane's 

55 from_above: bool # a line comes down into it (it's a parent of a commit above) 

56 through: list[tuple[int, int]] # lanes passing by (top to bottom) 

57 joining: list[tuple[int, int]] # lanes ending in it (from the top: a branch that started here) 

58 parents: list[tuple[int, int]] # lanes leaving it (to the bottom: its parents') 

59 width: int # lanes in use, above or below it 

60 

61 

62@dataclass 

63class Log: 

64 root: Path 

65 head: str # the commit checked out 

66 commits: list[LogCommit] 

67 rows: list[GraphRow] 

68 all_branches: bool 

69 limited: bool # more commits than the limit: the newest shown 

70 remote: str = "" # origin's address (the commits' web pages) 

71 total: int = 0 # the commits there are (more than those listed, if limited) 

72 

73 

74def parse_refs(decoration: str) -> list[Ref]: 

75 """The refs in ``%D`` (with ``--decorate=full``): "HEAD -> refs/heads/main, 

76 tag: refs/tags/v1, refs/remotes/origin/main". The branch checked out 

77 first, then the others, the remote ones, the tags.""" 

78 refs = [] 

79 for part in (p.strip() for p in decoration.split(",")): 

80 current = part.startswith("HEAD -> ") 

81 part = part.removeprefix("HEAD -> ") 

82 if part == "HEAD": 

83 refs.append(Ref("HEAD", HEAD, True)) # (detached) 

84 elif part.startswith("tag: refs/tags/"): 

85 refs.append(Ref(part.removeprefix("tag: refs/tags/"), TAG)) 

86 elif part.startswith("refs/heads/"): 

87 refs.append(Ref(part.removeprefix("refs/heads/"), BRANCH, current)) 

88 elif part.startswith("refs/remotes/") and not part.endswith("/HEAD"): 

89 refs.append(Ref(part.removeprefix("refs/remotes/"), REMOTE)) 

90 order = {HEAD: 0, BRANCH: 1, REMOTE: 2, TAG: 3} 

91 return sorted(refs, key=lambda r: (not r.current, order[r.kind])) 

92 

93 

94FIELDS = "%H%x1f%P%x1f%an%x1f%ae%x1f%at%x1f%D%x1f%s%x1e" 

95 

96 

97def parse_log(output: bytes) -> list[LogCommit]: 

98 """``git log --format=FIELDS --decorate=full`` read.""" 

99 commits = [] 

100 for record in output.decode("utf-8", "replace").split("\x1e"): 

101 parts = record.strip("\n").split("\x1f") 

102 if len(parts) != 7 or len(parts[0]) < 7: 

103 continue 

104 sha, parents, author, email, when, refs, subject = parts 

105 commits.append(LogCommit(sha, parents.split(), author, email, int(when) if when.isdigit() else 0, subject, 

106 parse_refs(refs))) 

107 return commits 

108 

109 

110def graph(commits: list[LogCommit]) -> list[GraphRow]: 

111 """Each commit's row of the graph (see the module's description).""" 

112 lanes: list[str | None] = [] # the commit each lane waits for 

113 colors: list[int] = [] 

114 next_color = 0 

115 rows = [] 

116 

117 def free_lane(avoid: int = -1) -> int: 

118 for i, waiting in enumerate(lanes): 

119 if waiting is None and i != avoid: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 return i 

121 lanes.append(None) 

122 colors.append(0) 

123 return len(lanes) - 1 

124 

125 for commit in commits: 

126 above = len(lanes) 

127 from_above = commit.sha in lanes 

128 if from_above: 

129 lane = lanes.index(commit.sha) 

130 else: 

131 lane = free_lane() 

132 lanes[lane], colors[lane] = commit.sha, next_color 

133 next_color += 1 

134 joining = [(i, colors[i]) for i, waiting in enumerate(lanes) if waiting == commit.sha and i != lane] 

135 through = [(i, colors[i]) for i, waiting in enumerate(lanes) if waiting is not None and waiting != commit.sha] 

136 for i, _color in joining: 

137 lanes[i] = None 

138 parents = [] 

139 if commit.parents: 

140 lanes[lane] = commit.parents[0] 

141 parents.append((lane, colors[lane])) 

142 for parent in commit.parents[1:]: 

143 if parent in lanes: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 i = lanes.index(parent) 

145 else: 

146 i = free_lane(avoid=lane) 

147 lanes[i], colors[i] = parent, next_color 

148 next_color += 1 

149 parents.append((i, colors[i])) 

150 else: 

151 lanes[lane] = None # (the first commit: nothing below) 

152 rows.append(GraphRow(lane, colors[lane], from_above, through, joining, parents, max(above, len(lanes)))) 

153 while lanes and lanes[-1] is None: 

154 lanes.pop() 

155 colors.pop() 

156 return rows 

157 

158 

159def read_log(path: Path, all_branches: bool = True, limit: int | None = MAX_COMMITS) -> Log | None: 

160 """The history of the repository ``path`` is in: every branch's (local, 

161 remote) and tag's, or only what's checked out (None: not in a 

162 repository, or no commits yet); the newest ``limit`` commits (None: 

163 all of them, to the first). Raises GitError if git can't read it.""" 

164 root = find_repository(path) 

165 if root is None: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true

166 return None 

167 code, head, _message = _git(root, "rev-parse", "--verify", "-q", "HEAD") 

168 if code != 0 or not head or not head.strip(): 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true

169 return None 

170 which = ["--branches", "--remotes", "--tags", "HEAD"] if all_branches else ["HEAD"] 

171 most = [] if limit is None else [f"--max-count={limit + 1}"] 

172 output = _check(root, "log", f"--format={FIELDS}", "--decorate=full", "--date-order", *most, *which, "--") 

173 commits = parse_log(output) 

174 limited = limit is not None and len(commits) > limit 

175 total = len(commits) 

176 if limited: 

177 commits = commits[:limit] 

178 total = int(_check(root, "rev-list", "--count", *which, "--").strip() or 0) # (how many more there are) 

179 _code, remote, _message = _git(root, "config", "--get", "remote.origin.url") 

180 return Log(root, head.decode().strip(), commits, graph(commits), all_branches, limited, 

181 (remote or b"").decode("utf-8", "replace").strip(), total) 

182 

183 

184def file_versions(root: Path, commit: LogCommit, change: FileChange) -> tuple[str | None, str | None]: 

185 """(before, after) a commit of a file it changed: the text in its first 

186 parent (None: added, or the first commit), and in it (None: deleted). 

187 BINARY: it isn't text, or it's huge.""" 

188 

189 def blob(spec: str) -> str | None: 

190 code, output, _message = _git(root, "cat-file", "blob", spec, limit=MAX_DIFF_BYTES) 

191 if code != 0: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true

192 return None 

193 return BINARY if output is None else _text(output) 

194 

195 parent = commit.parents[0] if commit.parents else None 

196 before = blob(f"{parent}:{change.old_path or change.path}") if parent and change.status != "A" else None 

197 after = blob(f"{commit.sha}:{change.path}") if change.status != "D" else None 

198 return before, after