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

1from pathlib import Path 

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

3 

4from git import Repo 

5from git.objects import Blob, Commit, IndexObject, Tree 

6from opal_common.paths import PathUtils 

7 

8 

9class VersionedNode: 

10 """A *versioned* file or a directory in a git repo. 

11 

12 VersionedNode is a base class for `VersionedFile` and 

13 `VersionedDirectory`. 

14 """ 

15 

16 def __init__(self, node: IndexObject, commit: Commit): 

17 self._node = node 

18 self._commit = commit 

19 self._repo: Repo = commit.repo 

20 

21 @property 

22 def repo(self) -> Repo: 

23 """The repo containing the versioned node.""" 

24 return self._repo 

25 

26 @property 

27 def commit(self) -> Commit: 

28 """The commit in which the node (blob, tree) is located.""" 

29 return self._commit 

30 

31 @property 

32 def version(self) -> str: 

33 """The hash (hex sha) of the node's parent commit.""" 

34 return self._commit.hexsha 

35 

36 @property 

37 def path(self) -> Path: 

38 """The relative path to the node (either file path or directory path), 

39 relative to the repo root.""" 

40 return Path(self._node.path) 

41 

42 

43class VersionedFile(VersionedNode): 

44 """Each instance of this class represents *one version* of a file (blob) in 

45 a git repo (the version of the file for a specific git commit).""" 

46 

47 def __init__(self, blob: Blob, commit: Commit): 

48 super().__init__(blob, commit) 

49 self._blob: Blob = blob 

50 

51 @property 

52 def blob(self) -> Blob: 

53 """The blob containing metadata for the file version.""" 

54 return self._blob 

55 

56 @property 

57 def stream(self) -> IO: 

58 """An io stream to the version of the file represented by that 

59 instance. 

60 

61 reading that stream will return the contents of the file for 

62 that specific version (commit). 

63 """ 

64 return self.blob.data_stream 

65 

66 def read_bytes(self) -> bytes: 

67 """Returns the contents of the file as a byte array (without 

68 encoding).""" 

69 return self.stream.read() 

70 

71 def read(self, encoding="utf-8") -> str: 

72 """Returns the contents of the file as a string, decoded according to 

73 the input `encoding`. 

74 

75 (by default, git usually encodes source files as utf-8). 

76 """ 

77 return self.read_bytes().decode(encoding=encoding) 

78 

79 

80class VersionedDirectory(VersionedNode): 

81 """Each instance of this class represents *one version* of a directory (git 

82 tree) in a git repo (the version of the directory for a specific git 

83 commit).""" 

84 

85 def __init__(self, directory: Tree, commit: Commit): 

86 super().__init__(directory, commit) 

87 self._dir: Tree = directory 

88 

89 @property 

90 def dir(self) -> Tree: 

91 """The git tree representing the metadata for that version of the 

92 directory. 

93 

94 i.e: one can get child directories (trees) and files (blobs) for 

95 the instance's version. 

96 """ 

97 return self._dir 

98 

99 

100NodeFilter = Callable[[VersionedNode], bool] 

101FileFilter = Callable[[VersionedFile], bool] 

102DirectoryFilter = Callable[[VersionedDirectory], bool] 

103 

104 

105def has_extension(f: VersionedFile, extensions: Optional[List[str]] = None) -> bool: 

106 """A filter on versioned files, filters only files with specific types 

107 (file extensions).""" 

108 if extensions is None: 

109 return True # no filter 

110 else: 

111 return f.path.suffix in extensions 

112 

113 

114def find_ignore_match( 

115 maybe_path: Path, bundle_ignore: Optional[List[str]] 

116) -> Optional[str]: 

117 """Determines the ignored glob path, if any, which matches the given file's 

118 path. 

119 

120 Returns the matched glob path rather than a binary decision of 

121 whether there is a match to enable better logging in the case of 

122 matched paths in manifests. 

123 """ 

124 if bundle_ignore is not None: 

125 return PathUtils.glob_style_match_path_to_list( 

126 Path(maybe_path).as_posix(), bundle_ignore 

127 ) 

128 return None 

129 

130 

131def is_under_directories(f: VersionedFile, directories: Set[Path]) -> bool: 

132 """A filter on versioned files, filters only files under certain 

133 directories in the repo.""" 

134 return PathUtils.is_child_of_directories(f.path, directories) 

135 

136 

137class CommitViewer: 

138 """This class allows us to view the repository files and directories from 

139 the perspective of a specific git commit (i.e: version). 

140 

141 i.e: if in the latest commit we removed a file called `a.txt`, we 

142 will see it while initializing CommitViewer with commit=HEAD~1, but 

143 we will not see `a.txt` if we initialize the CommitViewer with 

144 commit=HEAD. 

145 

146 The viewer also allows us to filter out certain paths of the commit 

147 tree. 

148 """ 

149 

150 def __init__(self, commit: Commit): 

151 """[summary] 

152 

153 Args: 

154 commit (Commit): the commit that defines the perspective (or lens) 

155 through which we look at the repo filesystem. i.e: the commit 

156 that defines the "checkout". 

157 """ 

158 self._repo: Repo = commit.repo 

159 self._commit = commit 

160 self._root = commit.tree 

161 

162 def __enter__(self): 

163 return self 

164 

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

166 pass 

167 

168 def nodes( 

169 self, predicate: Optional[NodeFilter] = None 

170 ) -> Generator[VersionedNode, None, None]: 

171 """A generator yielding all the nodes (files and directories) found in 

172 the repository for the current commit, after applying the filter. 

173 

174 Args: 

175 predicate (Optional[NodeFilter]): an optional predicate to filter only specific nodes. 

176 

177 Yields: 

178 the next node found (only for nodes passing the filter). 

179 """ 

180 nodes_generator = self._nodes_in_tree(self._root) 

181 if predicate is None: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true

182 return nodes_generator 

183 else: 

184 return filter(predicate, nodes_generator) 

185 

186 def files( 

187 self, predicate: Optional[FileFilter] = None 

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

189 """A generator yielding all the files found in the repository for the 

190 current commit, after applying the filter. 

191 

192 Args: 

193 filter (Optional[FileFilter]): an optional predicate to filter only specific files. 

194 

195 Yields: 

196 the next file found (only for files passing the filter). 

197 """ 

198 return ( 

199 node for node in self.nodes(predicate) if isinstance(node, VersionedFile) 

200 ) 

201 

202 def directories( 

203 self, predicate: Optional[DirectoryFilter] = None 

204 ) -> Generator[VersionedDirectory, None, None]: 

205 """A generator yielding all the directories found in the repository for 

206 the current commit, after applying the filter. 

207 

208 Args: 

209 filter (Optional[DirectoryFilter]): an optional predicate to filter only specific directories. 

210 

211 Yields: 

212 the next directory found (only for directories passing the filter). 

213 """ 

214 return filter( 

215 lambda node: isinstance(node, VersionedDirectory), 

216 self.nodes(predicate), 

217 ) 

218 

219 def get_node( 

220 self, path: Path, filterable_gen: Optional[Callable] = None 

221 ) -> Optional[VersionedNode]: 

222 """Returns the node in the given path, None if it doesn't exist.""" 

223 return next(self.nodes(lambda n: n.path == path), None) 

224 

225 def get_directory(self, path: Path) -> Optional[VersionedDirectory]: 

226 """Returns the directory in the given path, None if it doesn't 

227 exist.""" 

228 return next(self.directories(lambda n: n.path == path), None) 

229 

230 def get_file(self, path: Path) -> Optional[VersionedFile]: 

231 """Returns the file in the given path, None if it doesn't exist.""" 

232 return next(self.files(lambda n: n.path == path), None) 

233 

234 @property 

235 def paths(self) -> List[Path]: 

236 """Returns all the paths in the repo for the current commit (both files 

237 and directories)""" 

238 return [node.path for node in self.nodes()] 

239 

240 def exists(self, path: Path) -> bool: 

241 """Checks if a certain path exists in the repo in the current 

242 commit.""" 

243 return path in self.paths 

244 

245 def _nodes_in_tree(self, root: Tree) -> Generator[VersionedNode, None, None]: 

246 """A generator returning all the nodes (files and directories) under a 

247 certain git Tree (a versioned directory).""" 

248 # yield current directory 

249 yield VersionedDirectory(root, self._commit) 

250 # yield files under current directory 

251 for blob in root.blobs: 

252 yield VersionedFile(blob, self._commit) 

253 # yield subdirectories (and their children etc) under current directory 

254 for tree in root.trees: 254 ↛ 255line 254 didn't jump to line 255 because the loop on line 254 never started

255 yield from self._nodes_in_tree(tree)