Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/git_utils/diff_viewer.py: 20%

1from pathlib import Path 

2from typing import IO, Callable, Generator, List, Optional, Set 

3 

4from git import Repo 

5from git.diff import Diff, DiffIndex 

6from git.objects.commit import Commit 

7from opal_common.git_utils.commit_viewer import VersionedFile 

8from opal_common.paths import PathUtils 

9 

10DiffFilter = Callable[[Diff], bool] 

11PathFilter = Callable[[Path], bool] 

12 

13 

14def apply_filter( 

15 generator: Generator[Diff, None, None], filter: Optional[DiffFilter] = None 

16) -> Generator[Diff, None, None]: 

17 """Applies an optional filter on top of a Diff generator. 

18 

19 returns only the diffs yielded by the source generator that pass the 

20 filter. if no filter is provided, returns the same results as the 

21 source generator. 

22 """ 

23 if filter is None: 

24 yield from generator 

25 else: 

26 for diff in generator: 

27 if filter(diff): 

28 yield diff 

29 

30 

31def diffed_file_has_extension( 

32 diff: Diff, extensions: Optional[List[str]] = None 

33) -> bool: 

34 """Filter on git diffs, filters only diffs on files that has a certain 

35 extension/type. 

36 

37 if the file is renamed/added/removed, its enough that only one of 

38 its versions has the required extension. 

39 """ 

40 if extensions is None: 

41 return True # no filter 

42 

43 for path in [diff.a_path, diff.b_path]: 

44 if path is not None and Path(path).suffix in extensions: 

45 return True 

46 return False 

47 

48 

49def diffed_file_is_under_directories(diff: Diff, directories: Set[Path]) -> bool: 

50 """Filter on git diffs, filters only diffs on files that are located in 

51 certain directories. 

52 

53 if a file is renamed/added/removed, its enough that only one of its 

54 versions was located in one of the required directories. 

55 """ 

56 for path in [diff.a_path, diff.b_path]: 

57 if path is not None and PathUtils.is_child_of_directories( 

58 Path(path), directories 

59 ): 

60 return True 

61 return False 

62 

63 

64class DiffViewer: 

65 """This class allows us to view the changes made between two commits. 

66 

67 these two commits are not necessarily consecutive. 

68 """ 

69 

70 def __init__(self, old: Commit, new: Commit): 

71 """[summary] 

72 

73 Args: 

74 old (Commit): the older/earlier commit that defines the diff 

75 new (Commit): the newer/later commit that defines the diff 

76 """ 

77 if old.repo != new.repo: 

78 raise ValueError("you can only diff two commits from the same repo!") 

79 self._repo: Repo = old.repo 

80 self._old = old 

81 self._new = new 

82 self._diffs: DiffIndex = old.diff(new) 

83 

84 def __enter__(self): 

85 return self 

86 

87 def __exit__(self, exc_type, exc, tb): 

88 pass 

89 

90 def changes( 

91 self, filter: Optional[DiffFilter] = None 

92 ) -> Generator[Diff, None, None]: 

93 """A generator yielding all the diffs between the old commit and the 

94 new commit, after applying the filter. 

95 

96 Each diff (instance of `Diff`) is a change in one file, i.e: if the 

97 diff between the commits returned by git diff is as following: 

98 

99 Changes to be committed: 

100 (use "git restore --staged <file>..." to unstage) 

101 modified: server/main.py 

102 deleted: server/policy/publisher.py 

103 new file: server/publisher.py 

104 

105 then 3 Diffs will be returned (one modified, one deleted, one added). 

106 

107 Args: 

108 filter (Optional[DiffFilter]): an optional predicate to filter only specific diffs. 

109 

110 Yields: 

111 the next diff found (only for diffs passing the filter). 

112 """ 

113 for diff in self._diffs: 

114 if filter is None: 

115 yield diff 

116 elif filter(diff): 

117 yield diff 

118 

119 def added(self, filter: Optional[DiffFilter] = None) -> Generator[Diff, None, None]: 

120 """A generator yielding all the diffs between the old commit and the 

121 new commit, that are of type "new file" (i.e: added), after applying 

122 the filter. 

123 

124 @see `changes()` 

125 """ 

126 diff_generator = self._diffs.iter_change_type("A") 

127 yield from apply_filter(diff_generator, filter) 

128 

129 def deleted( 

130 self, filter: Optional[DiffFilter] = None 

131 ) -> Generator[Diff, None, None]: 

132 """A generator yielding all the diffs between the old commit and the 

133 new commit, that are of type "deleted", after applying the filter. 

134 

135 @see `changes()` 

136 """ 

137 diff_generator = self._diffs.iter_change_type("D") 

138 yield from apply_filter(diff_generator, filter) 

139 

140 def renamed( 

141 self, filter: Optional[DiffFilter] = None 

142 ) -> Generator[Diff, None, None]: 

143 """A generator yielding all the diffs between the old commit and the 

144 new commit, that are of type "renamed", after applying the filter. 

145 

146 @see `changes()` 

147 """ 

148 diff_generator = self._diffs.iter_change_type("R") 

149 yield from apply_filter(diff_generator, filter) 

150 

151 def modified( 

152 self, filter: Optional[DiffFilter] = None 

153 ) -> Generator[Diff, None, None]: 

154 """A generator yielding all the diffs between the old commit and the 

155 new commit, that are of type "modified", after applying the filter. 

156 

157 @see `changes()` 

158 """ 

159 diff_generator = self._diffs.iter_change_type("M") 

160 yield from apply_filter(diff_generator, filter) 

161 

162 def added_files( 

163 self, filter: Optional[DiffFilter] = None 

164 ) -> Generator[VersionedFile, None, None]: 

165 """A generator yielding the new version (blob) of files that were added 

166 (or renamed, meaning a new file was added under the new name) in the 

167 diff, between the old and new commits. 

168 

169 In both cases, a new file that has not existed before was added 

170 to the repo. 

171 """ 

172 for diff in self.added(filter): 

173 yield VersionedFile(diff.b_blob, self._new) 

174 

175 for diff in self.renamed(filter): 

176 yield VersionedFile(diff.b_blob, self._new) 

177 

178 def deleted_files( 

179 self, filter: Optional[DiffFilter] = None 

180 ) -> Generator[VersionedFile, None, None]: 

181 """A generator yielding the old version (blob) of the files that were 

182 removed (or renamed, meaning the file under the old name was removed) 

183 in the diff between the old and new commits. 

184 

185 In both cases, a file is removed from the repo. 

186 """ 

187 for diff in self.deleted(filter): 

188 yield VersionedFile(diff.a_blob, self._old) 

189 

190 for diff in self.renamed(filter): 

191 yield VersionedFile(diff.a_blob, self._old) 

192 

193 def modified_files( 

194 self, filter: Optional[DiffFilter] = None 

195 ) -> Generator[VersionedFile, None, None]: 

196 """A generator yielding the new version (blob) of files that were 

197 changed (modified) in the diff between the old and new commit.""" 

198 for diff in self.modified(filter): 

199 yield VersionedFile(diff.b_blob, self._new) 

200 

201 def added_or_modified_files( 

202 self, filter: Optional[DiffFilter] = None 

203 ) -> Generator[VersionedFile, None, None]: 

204 """A shortcut generator yield both `added_files()` and 

205 `modified_files()`.""" 

206 yield from self.added_files(filter) 

207 yield from self.modified_files(filter) 

208 

209 def affected_paths(self, filter: Optional[PathFilter] = None) -> Set[Path]: 

210 """Returns the set of paths of all files that were affected in the diff 

211 between the old and new commits. 

212 

213 only file paths are returned (and not directories). 

214 """ 

215 paths = set() 

216 for diff in self._diffs: 

217 diff: Diff 

218 for str_path in [diff.a_path, diff.b_path]: 

219 if str_path is not None: 

220 path = Path(str_path) 

221 if filter is None: 

222 paths.add(path) 

223 elif filter(path): 

224 paths.add(path) 

225 return paths