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

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 public 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 BinaryIO, Optional, Union 

27 

28from .objects import ZERO_SHA, format_timezone, parse_timezone 

29 

30Entry = collections.namedtuple( 

31 "Entry", 

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

33) 

34 

35 

36def format_reflog_line( 

37 old_sha: Optional[bytes], 

38 new_sha: bytes, 

39 committer: bytes, 

40 timestamp: Union[int, float], 

41 timezone: int, 

42 message: bytes, 

43) -> bytes: 

44 """Generate a single reflog line. 

45 

46 Args: 

47 old_sha: Old Commit SHA 

48 new_sha: New Commit SHA 

49 committer: Committer name and e-mail 

50 timestamp: Timestamp 

51 timezone: Timezone 

52 message: Message 

53 """ 

54 if old_sha is None: 

55 old_sha = ZERO_SHA 

56 return ( 

57 old_sha 

58 + b" " 

59 + new_sha 

60 + b" " 

61 + committer 

62 + b" " 

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

64 + b" " 

65 + format_timezone(timezone) 

66 + b"\t" 

67 + message 

68 ) 

69 

70 

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

72 """Parse a reflog line. 

73 

74 Args: 

75 line: Line to parse 

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

77 message) 

78 """ 

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

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

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

82 return Entry( 

83 old_sha, 

84 new_sha, 

85 committer, 

86 int(timestamp_str), 

87 parse_timezone(timezone_str)[0], 

88 message, 

89 ) 

90 

91 

92def read_reflog(f: BinaryIO) -> Generator[Entry, None, None]: 

93 """Read reflog. 

94 

95 Args: 

96 f: File-like object 

97 Returns: Iterator over Entry objects 

98 """ 

99 for line in f: 

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

101 

102 

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

104 """Drop the specified reflog entry. 

105 

106 Args: 

107 f: File-like object 

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

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

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

111 """ 

112 if index < 0: 

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

114 

115 log = [] 

116 offset = f.tell() 

117 for line in f: 

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

119 offset = f.tell() 

120 

121 inverse_index = len(log) - index - 1 

122 write_offset = log[inverse_index][0] 

123 f.seek(write_offset) 

124 

125 if index == 0: 

126 f.truncate() 

127 return 

128 

129 del log[inverse_index] 

130 if rewrite and index > 0 and log: 

131 if inverse_index == 0: 

132 previous_new = ZERO_SHA 

133 else: 

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

135 offset, entry = log[inverse_index] 

136 log[inverse_index] = ( 

137 offset, 

138 Entry( 

139 previous_new, 

140 entry.new_sha, 

141 entry.committer, 

142 entry.timestamp, 

143 entry.timezone, 

144 entry.message, 

145 ), 

146 ) 

147 

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

149 f.write( 

150 format_reflog_line( 

151 entry.old_sha, 

152 entry.new_sha, 

153 entry.committer, 

154 entry.timestamp, 

155 entry.timezone, 

156 entry.message, 

157 ) 

158 ) 

159 f.truncate() 

160 

161 

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

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

164 

165 Args: 

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

167 

168 Yields: 

169 Reference names (as bytes) that have reflogs 

170 """ 

171 import os 

172 from pathlib import Path 

173 

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

175 return 

176 

177 logs_path = Path(logs_dir) 

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

179 if log_file.is_file(): 

180 # Get the ref name by removing the logs_dir prefix 

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

182 # Convert path separators to / for refs 

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

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