Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/refs/log.py: 48%

1# This module is part of GitPython and is released under the 

2# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ 

3 

4__all__ = ["RefLog", "RefLogEntry"] 

5 

6from mmap import mmap 

7import re 

8import time as _time 

9 

10from git.compat import defenc 

11from git.objects.util import ( 

12 Serializable, 

13 altz_to_utctz_str, 

14 parse_date, 

15) 

16from git.util import ( 

17 Actor, 

18 LockedFD, 

19 LockFile, 

20 assure_directory_exists, 

21 bin_to_hex, 

22 file_contents_ro_filepath, 

23 to_native_path, 

24) 

25 

26# typing ------------------------------------------------------------------ 

27 

28from typing import Iterator, List, Tuple, TYPE_CHECKING, Union 

29 

30from git.types import PathLike 

31 

32if TYPE_CHECKING: 

33 from io import BytesIO 

34 

35 from git.config import GitConfigParser, SectionConstraint 

36 from git.refs import SymbolicReference 

37 

38# ------------------------------------------------------------------------------ 

39 

40 

41class RefLogEntry(Tuple[str, str, Actor, Tuple[int, int], str]): 

42 """Named tuple allowing easy access to the revlog data fields.""" 

43 

44 _re_hexsha_only = re.compile(r"^[0-9A-Fa-f]{40}$") 

45 

46 __slots__ = () 

47 

48 def __repr__(self) -> str: 

49 """Representation of ourselves in git reflog format.""" 

50 return self.format() 

51 

52 def format(self) -> str: 

53 """:return: A string suitable to be placed in a reflog file.""" 

54 act = self.actor 

55 time = self.time 

56 return "{} {} {} <{}> {!s} {}\t{}\n".format( 

57 self.oldhexsha, 

58 self.newhexsha, 

59 act.name, 

60 act.email, 

61 time[0], 

62 altz_to_utctz_str(time[1]), 

63 self.message, 

64 ) 

65 

66 @property 

67 def oldhexsha(self) -> str: 

68 """The hexsha to the commit the ref pointed to before the change.""" 

69 return self[0] 

70 

71 @property 

72 def newhexsha(self) -> str: 

73 """The hexsha to the commit the ref now points to, after the change.""" 

74 return self[1] 

75 

76 @property 

77 def actor(self) -> Actor: 

78 """Actor instance, providing access.""" 

79 return self[2] 

80 

81 @property 

82 def time(self) -> Tuple[int, int]: 

83 """Time as tuple: 

84 

85 * [0] = ``int(time)`` 

86 * [1] = ``int(timezone_offset)`` in :attr:`time.altzone` format 

87 """ 

88 return self[3] 

89 

90 @property 

91 def message(self) -> str: 

92 """Message describing the operation that acted on the reference.""" 

93 return self[4] 

94 

95 @classmethod 

96 def new( 

97 cls, 

98 oldhexsha: str, 

99 newhexsha: str, 

100 actor: Actor, 

101 time: int, 

102 tz_offset: int, 

103 message: str, 

104 ) -> "RefLogEntry": # skipcq: PYL-W0621 

105 """:return: New instance of a :class:`RefLogEntry`""" 

106 if not isinstance(actor, Actor): 

107 raise ValueError("Need actor instance, got %s" % actor) 

108 # END check types 

109 return RefLogEntry((oldhexsha, newhexsha, actor, (time, tz_offset), message)) 

110 

111 @classmethod 

112 def from_line(cls, line: bytes) -> "RefLogEntry": 

113 """:return: New :class:`RefLogEntry` instance from the given revlog line. 

114 

115 :param line: 

116 Line bytes without trailing newline 

117 

118 :raise ValueError: 

119 If `line` could not be parsed. 

120 """ 

121 line_str = line.decode(defenc) 

122 fields = line_str.split("\t", 1) 

123 if len(fields) == 1: 

124 info, msg = fields[0], None 

125 elif len(fields) == 2: 

126 info, msg = fields 

127 else: 

128 raise ValueError("Line must have up to two TAB-separated fields. Got %s" % repr(line_str)) 

129 # END handle first split 

130 

131 oldhexsha = info[:40] 

132 newhexsha = info[41:81] 

133 for hexsha in (oldhexsha, newhexsha): 

134 if not cls._re_hexsha_only.match(hexsha): 

135 raise ValueError("Invalid hexsha: %r" % (hexsha,)) 

136 # END if hexsha re doesn't match 

137 # END for each hexsha 

138 

139 email_end = info.find(">", 82) 

140 if email_end == -1: 

141 raise ValueError("Missing token: >") 

142 # END handle missing end brace 

143 

144 actor = Actor._from_string(info[82 : email_end + 1]) 

145 time, tz_offset = parse_date(info[email_end + 2 :]) # skipcq: PYL-W0621 

146 

147 return RefLogEntry((oldhexsha, newhexsha, actor, (time, tz_offset), msg)) # type: ignore [arg-type] 

148 

149 

150class RefLog(List[RefLogEntry], Serializable): 

151 R"""A reflog contains :class:`RefLogEntry`\s, each of which defines a certain state 

152 of the head in question. Custom query methods allow to retrieve log entries by date 

153 or by other criteria. 

154 

155 Reflog entries are ordered. The first added entry is first in the list. The last 

156 entry, i.e. the last change of the head or reference, is last in the list. 

157 """ 

158 

159 __slots__ = ("_path",) 

160 

161 def __new__(cls, filepath: Union[PathLike, None] = None) -> "RefLog": 

162 inst = super().__new__(cls) 

163 return inst 

164 

165 def __init__(self, filepath: Union[PathLike, None] = None) -> None: 

166 """Initialize this instance with an optional filepath, from which we will 

167 initialize our data. The path is also used to write changes back using the 

168 :meth:`write` method.""" 

169 self._path = filepath 

170 if filepath is not None: 

171 self._read_from_file() 

172 # END handle filepath 

173 

174 def _read_from_file(self) -> None: 

175 try: 

176 fmap = file_contents_ro_filepath(self._path, stream=True, allow_mmap=True) 

177 except OSError: 

178 # It is possible and allowed that the file doesn't exist! 

179 return 

180 # END handle invalid log 

181 

182 try: 

183 self._deserialize(fmap) 

184 finally: 

185 fmap.close() 

186 # END handle closing of handle 

187 

188 # { Interface 

189 

190 @classmethod 

191 def from_file(cls, filepath: PathLike) -> "RefLog": 

192 """ 

193 :return: 

194 A new :class:`RefLog` instance containing all entries from the reflog at the 

195 given `filepath`. 

196 

197 :param filepath: 

198 Path to reflog. 

199 

200 :raise ValueError: 

201 If the file could not be read or was corrupted in some way. 

202 """ 

203 return cls(filepath) 

204 

205 @classmethod 

206 def path(cls, ref: "SymbolicReference") -> str: 

207 """ 

208 :return: 

209 String to absolute path at which the reflog of the given ref instance would 

210 be found. The path is not guaranteed to point to a valid file though. 

211 

212 :param ref: 

213 :class:`~git.refs.symbolic.SymbolicReference` instance 

214 

215 :raise ValueError: 

216 If `ref.path` is invalid or escapes the repository's reflog directory. 

217 """ 

218 return to_native_path(ref._get_validated_reflog_path(ref.repo, ref.path)) 

219 

220 @classmethod 

221 def iter_entries(cls, stream: Union[str, "BytesIO", mmap]) -> Iterator[RefLogEntry]: 

222 """ 

223 :return: 

224 Iterator yielding :class:`RefLogEntry` instances, one for each line read 

225 from the given stream. 

226 

227 :param stream: 

228 File-like object containing the revlog in its native format or string 

229 instance pointing to a file to read. 

230 """ 

231 new_entry = RefLogEntry.from_line 

232 if isinstance(stream, str): 

233 # Default args return mmap since Python 3. 

234 _stream = file_contents_ro_filepath(stream) 

235 assert isinstance(_stream, mmap) 

236 else: 

237 _stream = stream 

238 # END handle stream type 

239 while True: 

240 line = _stream.readline() 

241 if not line: 

242 return 

243 yield new_entry(line.strip()) 

244 # END endless loop 

245 

246 @classmethod 

247 def entry_at(cls, filepath: PathLike, index: int) -> "RefLogEntry": 

248 """ 

249 :return: 

250 :class:`RefLogEntry` at the given index. 

251 

252 :param filepath: 

253 Full path to the index file from which to read the entry. 

254 

255 :param index: 

256 Python list compatible index, i.e. it may be negative to specify an entry 

257 counted from the end of the list. 

258 

259 :raise IndexError: 

260 If the entry didn't exist. 

261 

262 :note: 

263 This method is faster as it only parses the entry at index, skipping all 

264 other lines. Nonetheless, the whole file has to be read if the index is 

265 negative. 

266 """ 

267 with open(filepath, "rb") as fp: 

268 if index < 0: 

269 return RefLogEntry.from_line(fp.readlines()[index].strip()) 

270 # Read until index is reached. 

271 

272 line = b"" 

273 for i in range(index + 1): 

274 line = fp.readline() 

275 if not line: 

276 raise IndexError(f"Index file ended at line {i + 1}, before given index was reached") 

277 # END abort on eof 

278 # END handle runup 

279 

280 return RefLogEntry.from_line(line.strip()) 

281 # END handle index 

282 

283 def to_file(self, filepath: PathLike) -> None: 

284 """Write the contents of the reflog instance to a file at the given filepath. 

285 

286 :param filepath: 

287 Path to file. Parent directories are assumed to exist. 

288 """ 

289 lfd = LockedFD(filepath) 

290 assure_directory_exists(filepath, is_file=True) 

291 

292 fp = lfd.open(write=True, stream=True) 

293 try: 

294 self._serialize(fp) 

295 lfd.commit() 

296 except BaseException: 

297 lfd.rollback() 

298 raise 

299 # END handle change 

300 

301 @classmethod 

302 def append_entry( 

303 cls, 

304 config_reader: Union[Actor, "GitConfigParser", "SectionConstraint", None], 

305 filepath: PathLike, 

306 oldbinsha: bytes, 

307 newbinsha: bytes, 

308 message: str, 

309 write: bool = True, 

310 ) -> "RefLogEntry": 

311 """Append a new log entry to the revlog at filepath. 

312 

313 :param config_reader: 

314 Configuration reader of the repository - used to obtain user information. 

315 May also be an :class:`~git.util.Actor` instance identifying the committer 

316 directly or ``None``. 

317 

318 :param filepath: 

319 Full path to the log file. 

320 

321 :param oldbinsha: 

322 Binary sha of the previous commit. 

323 

324 :param newbinsha: 

325 Binary sha of the current commit. 

326 

327 :param message: 

328 Message describing the change to the reference. 

329 

330 :param write: 

331 If ``True``, the changes will be written right away. 

332 Otherwise the change will not be written. 

333 

334 :return: 

335 :class:`RefLogEntry` objects which was appended to the log. 

336 

337 :note: 

338 As we are append-only, concurrent access is not a problem as we do not 

339 interfere with readers. 

340 """ 

341 

342 if len(oldbinsha) != 20 or len(newbinsha) != 20: 

343 raise ValueError("Shas need to be given in binary format") 

344 # END handle sha type 

345 assure_directory_exists(filepath, is_file=True) 

346 first_line = message.split("\n")[0] 

347 if isinstance(config_reader, Actor): 

348 committer = config_reader # mypy thinks this is Actor | Gitconfigparser, but why? 

349 else: 

350 committer = Actor.committer(config_reader) 

351 entry = RefLogEntry( 

352 ( 

353 bin_to_hex(oldbinsha).decode("ascii"), 

354 bin_to_hex(newbinsha).decode("ascii"), 

355 committer, 

356 (int(_time.time()), _time.altzone), 

357 first_line, 

358 ) 

359 ) 

360 

361 if write: 

362 lf = LockFile(filepath) 

363 lf._obtain_lock_or_raise() 

364 fd = open(filepath, "ab") 

365 try: 

366 fd.write(entry.format().encode(defenc)) 

367 finally: 

368 fd.close() 

369 lf._release_lock() 

370 # END handle write operation 

371 return entry 

372 

373 def write(self) -> "RefLog": 

374 """Write this instance's data to the file we are originating from. 

375 

376 :return: 

377 self 

378 """ 

379 if self._path is None: 

380 raise ValueError("Instance was not initialized with a path, use to_file(...) instead") 

381 # END assert path 

382 self.to_file(self._path) 

383 return self 

384 

385 # } END interface 

386 

387 # { Serializable Interface 

388 

389 def _serialize(self, stream: "BytesIO") -> "RefLog": 

390 write = stream.write 

391 

392 # Write all entries. 

393 for e in self: 

394 write(e.format().encode(defenc)) 

395 # END for each entry 

396 return self 

397 

398 def _deserialize(self, stream: "BytesIO") -> "RefLog": 

399 self.extend(self.iter_entries(stream)) 

400 return self 

401 

402 # } END serializable interface