Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/reflog.py: 20%

1# reflog.py -- Parsing and writing reflog files 

2# Copyright (C) 2015 Jelmer Vernooij and others. 

3# 

4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later 

5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU 

6# General Public License as published by the Free Software Foundation; version 2.0 

7# or (at your option) any later version. You can redistribute it and/or 

8# modify it under the terms of either of these two licenses. 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16# You should have received a copy of the licenses; if not, see 

17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License 

18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache 

19# License, Version 2.0. 

20# 

21 

22"""Utilities for reading and generating reflogs.""" 

23 

24__all__ = [ 

25 "Entry", 

26 "drop_reflog_entry", 

27 "expire_reflog", 

28 "format_reflog_line", 

29 "iter_reflogs", 

30 "parse_reflog_line", 

31 "parse_reflog_spec", 

32 "read_reflog", 

33] 

34 

35import collections 

36import os 

37from collections.abc import Callable, Generator 

38from pathlib import Path 

39from typing import IO, BinaryIO 

40 

41from .file import _GitFile 

42from .objects import ZERO_SHA, format_timezone, parse_timezone 

43 

44Entry = collections.namedtuple( 

45 "Entry", 

46 ["old_sha", "new_sha", "committer", "timestamp", "timezone", "message"], 

47) 

48 

49 

50def parse_reflog_spec(refspec: str | bytes) -> tuple[bytes, int]: 

51 """Parse a reflog specification like 'HEAD@{1}' or 'refs/heads/master@{2}'. 

52 

53 Args: 

54 refspec: Reflog specification (e.g., 'HEAD@{1}', 'master@{0}') 

55 

56 Returns: 

57 Tuple of (ref_name, index) where index is in Git reflog order (0 = newest) 

58 

59 Raises: 

60 ValueError: If the refspec is not a valid reflog specification 

61 """ 

62 if isinstance(refspec, str): 

63 refspec = refspec.encode("utf-8") 

64 

65 if b"@{" not in refspec: 

66 raise ValueError( 

67 f"Invalid reflog spec: {refspec!r}. Expected format: ref@{{n}}" 

68 ) 

69 

70 ref, rest = refspec.split(b"@{", 1) 

71 if not rest.endswith(b"}"): 

72 raise ValueError( 

73 f"Invalid reflog spec: {refspec!r}. Expected format: ref@{{n}}" 

74 ) 

75 

76 index_str = rest[:-1] 

77 if not index_str.isdigit(): 

78 raise ValueError( 

79 f"Invalid reflog index: {index_str!r}. Expected integer in ref@{{n}}" 

80 ) 

81 

82 # Use HEAD if no ref specified (e.g., "@{1}") 

83 if not ref: 

84 ref = b"HEAD" 

85 

86 return ref, int(index_str) 

87 

88 

89def format_reflog_line( 

90 old_sha: bytes | None, 

91 new_sha: bytes, 

92 committer: bytes, 

93 timestamp: int | float, 

94 timezone: int, 

95 message: bytes, 

96) -> bytes: 

97 """Generate a single reflog line. 

98 

99 Args: 

100 old_sha: Old Commit SHA 

101 new_sha: New Commit SHA 

102 committer: Committer name and e-mail 

103 timestamp: Timestamp 

104 timezone: Timezone 

105 message: Message 

106 """ 

107 if old_sha is None: 

108 old_sha = ZERO_SHA 

109 # A reflog entry is a single line and the message is the last, tab-separated 

110 # field. Collapse embedded whitespace (in particular newlines) to single 

111 # spaces so a multi-line or crafted message cannot split into extra physical 

112 # lines and forge additional reflog entries. Matches Git's copy_reflog_msg. 

113 message = b" ".join(message.split()) 

114 return ( 

115 old_sha 

116 + b" " 

117 + new_sha 

118 + b" " 

119 + committer 

120 + b" " 

121 + str(int(timestamp)).encode("ascii") 

122 + b" " 

123 + format_timezone(timezone) 

124 + b"\t" 

125 + message 

126 ) 

127 

128 

129def parse_reflog_line(line: bytes) -> Entry: 

130 """Parse a reflog line. 

131 

132 Args: 

133 line: Line to parse 

134 Returns: Tuple of (old_sha, new_sha, committer, timestamp, timezone, 

135 message) 

136 """ 

137 (begin, message) = line.split(b"\t", 1) 

138 (old_sha, new_sha, rest) = begin.split(b" ", 2) 

139 (committer, timestamp_str, timezone_str) = rest.rsplit(b" ", 2) 

140 return Entry( 

141 old_sha, 

142 new_sha, 

143 committer, 

144 int(timestamp_str), 

145 parse_timezone(timezone_str)[0], 

146 message, 

147 ) 

148 

149 

150def read_reflog( 

151 f: BinaryIO | IO[bytes] | _GitFile, 

152) -> Generator[Entry, None, None]: 

153 """Read reflog. 

154 

155 Args: 

156 f: File-like object 

157 Returns: Iterator over Entry objects 

158 """ 

159 for line in f: 

160 yield parse_reflog_line(line.rstrip(b"\n")) 

161 

162 

163def drop_reflog_entry(f: BinaryIO, index: int, rewrite: bool = False) -> None: 

164 """Drop the specified reflog entry. 

165 

166 Args: 

167 f: File-like object 

168 index: Reflog entry index (in Git reflog reverse 0-indexed order) 

169 rewrite: If a reflog entry's predecessor is removed, set its 

170 old SHA to the new SHA of the entry that now precedes it 

171 """ 

172 if index < 0: 

173 raise ValueError(f"Invalid reflog index {index}") 

174 

175 log = [] 

176 offset = f.tell() 

177 for line in f: 

178 log.append((offset, parse_reflog_line(line.rstrip(b"\n")))) 

179 offset = f.tell() 

180 

181 inverse_index = len(log) - index - 1 

182 write_offset = log[inverse_index][0] 

183 f.seek(write_offset) 

184 

185 if index == 0: 

186 f.truncate() 

187 return 

188 

189 del log[inverse_index] 

190 if rewrite and index > 0 and log: 

191 if inverse_index == 0: 

192 previous_new = ZERO_SHA 

193 else: 

194 previous_new = log[inverse_index - 1][1].new_sha 

195 offset, entry = log[inverse_index] 

196 log[inverse_index] = ( 

197 offset, 

198 Entry( 

199 previous_new, 

200 entry.new_sha, 

201 entry.committer, 

202 entry.timestamp, 

203 entry.timezone, 

204 entry.message, 

205 ), 

206 ) 

207 

208 for _, entry in log[inverse_index:]: 

209 f.write( 

210 format_reflog_line( 

211 entry.old_sha, 

212 entry.new_sha, 

213 entry.committer, 

214 entry.timestamp, 

215 entry.timezone, 

216 entry.message, 

217 ) 

218 + b"\n" 

219 ) 

220 f.truncate() 

221 

222 

223def expire_reflog( 

224 f: BinaryIO, 

225 expire_time: int | None = None, 

226 expire_unreachable_time: int | None = None, 

227 reachable_checker: Callable[[bytes], bool] | None = None, 

228) -> int: 

229 """Expire reflog entries based on age and reachability. 

230 

231 Args: 

232 f: File-like object for the reflog 

233 expire_time: Expire entries older than this timestamp (seconds since epoch). 

234 If None, entries are not expired based on age alone. 

235 expire_unreachable_time: Expire unreachable entries older than this 

236 timestamp. If None, unreachable entries are not expired. 

237 reachable_checker: Optional callable that takes a SHA and returns True 

238 if the commit is reachable. If None, all entries are considered 

239 reachable. 

240 

241 Returns: 

242 Number of entries expired 

243 """ 

244 if expire_time is None and expire_unreachable_time is None: 

245 return 0 

246 

247 entries = [] 

248 offset = f.tell() 

249 for line in f: 

250 entries.append((offset, parse_reflog_line(line.rstrip(b"\n")))) 

251 offset = f.tell() 

252 

253 # Filter entries that should be kept 

254 kept_entries = [] 

255 expired_count = 0 

256 

257 for offset, entry in entries: 

258 should_expire = False 

259 

260 # Check if entry is reachable 

261 is_reachable = True 

262 if reachable_checker is not None: 

263 is_reachable = reachable_checker(entry.new_sha) 

264 

265 # Apply expiration rules 

266 # Check the appropriate expiration time based on reachability 

267 if is_reachable: 

268 if expire_time is not None and entry.timestamp < expire_time: 

269 should_expire = True 

270 else: 

271 if ( 

272 expire_unreachable_time is not None 

273 and entry.timestamp < expire_unreachable_time 

274 ): 

275 should_expire = True 

276 

277 if should_expire: 

278 expired_count += 1 

279 else: 

280 kept_entries.append((offset, entry)) 

281 

282 # Write back the kept entries 

283 if expired_count > 0: 

284 f.seek(0) 

285 for _, entry in kept_entries: 

286 f.write( 

287 format_reflog_line( 

288 entry.old_sha, 

289 entry.new_sha, 

290 entry.committer, 

291 entry.timestamp, 

292 entry.timezone, 

293 entry.message, 

294 ) 

295 + b"\n" 

296 ) 

297 f.truncate() 

298 

299 return expired_count 

300 

301 

302def iter_reflogs(logs_dir: str) -> Generator[bytes, None, None]: 

303 """Iterate over all reflogs in a repository. 

304 

305 Args: 

306 logs_dir: Path to the logs directory (e.g., .git/logs) 

307 

308 Yields: 

309 Reference names (as bytes) that have reflogs 

310 """ 

311 if not os.path.exists(logs_dir): 

312 return 

313 

314 logs_path = Path(logs_dir) 

315 for log_file in logs_path.rglob("*"): 

316 if log_file.is_file(): 

317 # Get the ref name by removing the logs_dir prefix 

318 ref_name = str(log_file.relative_to(logs_path)) 

319 # Convert path separators to / for refs 

320 ref_name = ref_name.replace(os.sep, "/") 

321 yield ref_name.encode("utf-8")