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

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 

24import collections 

25from collections.abc import Generator 

26from typing import IO, BinaryIO, Optional, Union 

27 

28from .file import _GitFile 

29from .objects import ZERO_SHA, format_timezone, parse_timezone 

30 

31Entry = collections.namedtuple( 

32 "Entry", 

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

34) 

35 

36 

37def format_reflog_line( 

38 old_sha: Optional[bytes], 

39 new_sha: bytes, 

40 committer: bytes, 

41 timestamp: Union[int, float], 

42 timezone: int, 

43 message: bytes, 

44) -> bytes: 

45 """Generate a single reflog line. 

46 

47 Args: 

48 old_sha: Old Commit SHA 

49 new_sha: New Commit SHA 

50 committer: Committer name and e-mail 

51 timestamp: Timestamp 

52 timezone: Timezone 

53 message: Message 

54 """ 

55 if old_sha is None: 

56 old_sha = ZERO_SHA 

57 return ( 

58 old_sha 

59 + b" " 

60 + new_sha 

61 + b" " 

62 + committer 

63 + b" " 

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

65 + b" " 

66 + format_timezone(timezone) 

67 + b"\t" 

68 + message 

69 ) 

70 

71 

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

73 """Parse a reflog line. 

74 

75 Args: 

76 line: Line to parse 

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

78 message) 

79 """ 

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

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

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

83 return Entry( 

84 old_sha, 

85 new_sha, 

86 committer, 

87 int(timestamp_str), 

88 parse_timezone(timezone_str)[0], 

89 message, 

90 ) 

91 

92 

93def read_reflog( 

94 f: Union[BinaryIO, IO[bytes], _GitFile], 

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

96 """Read reflog. 

97 

98 Args: 

99 f: File-like object 

100 Returns: Iterator over Entry objects 

101 """ 

102 for line in f: 

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

104 

105 

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

107 """Drop the specified reflog entry. 

108 

109 Args: 

110 f: File-like object 

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

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

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

114 """ 

115 if index < 0: 

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

117 

118 log = [] 

119 offset = f.tell() 

120 for line in f: 

121 log.append((offset, parse_reflog_line(line))) 

122 offset = f.tell() 

123 

124 inverse_index = len(log) - index - 1 

125 write_offset = log[inverse_index][0] 

126 f.seek(write_offset) 

127 

128 if index == 0: 

129 f.truncate() 

130 return 

131 

132 del log[inverse_index] 

133 if rewrite and index > 0 and log: 

134 if inverse_index == 0: 

135 previous_new = ZERO_SHA 

136 else: 

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

138 offset, entry = log[inverse_index] 

139 log[inverse_index] = ( 

140 offset, 

141 Entry( 

142 previous_new, 

143 entry.new_sha, 

144 entry.committer, 

145 entry.timestamp, 

146 entry.timezone, 

147 entry.message, 

148 ), 

149 ) 

150 

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

152 f.write( 

153 format_reflog_line( 

154 entry.old_sha, 

155 entry.new_sha, 

156 entry.committer, 

157 entry.timestamp, 

158 entry.timezone, 

159 entry.message, 

160 ) 

161 ) 

162 f.truncate() 

163 

164 

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

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

167 

168 Args: 

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

170 

171 Yields: 

172 Reference names (as bytes) that have reflogs 

173 """ 

174 import os 

175 from pathlib import Path 

176 

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

178 return 

179 

180 logs_path = Path(logs_dir) 

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

182 if log_file.is_file(): 

183 # Get the ref name by removing the logs_dir prefix 

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

185 # Convert path separators to / for refs 

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

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