1""" 

2Checkout command. 

3 

4Clone a repository into the given directory on the host machine. 

5""" 

6import os.path 

7import shutil 

8import sys 

9from pathlib import Path 

10from textwrap import dedent 

11from typing import Dict, List, Type 

12 

13import git 

14import taxonomy 

15from errors import (DppGitApplyPatchError, DppGitCheckoutError, DppGitCheckoutInvalidRepositoryError, DppGitCloneError, 

16 DppGitError, DppGitPatchNotAppliedError, DppGitSubmoduleInitError, DppGitWorktreeError) 

17from message import message 

18from processor.core.argparser import create_common_vcs_parser 

19from processor.core.command import Command 

20from processor.core.data import Project 

21from processor.core.docker import Docker 

22 

23 

24def _git_clone(path: Path, metadata: taxonomy.MetaData) -> git.Repo: 

25 """ 

26 Initialize repository or clone a new one if it does not exist. 

27 """ 

28 try: 

29 repo = git.Repo(str(path)) 

30 except git.NoSuchPathError: 

31 if not path.parent.exists(): 

32 path.parent.mkdir(parents=True, exist_ok=True) 

33 message.info(__name__, f"cloning {metadata.name} into {str(path)}") 

34 try: 

35 repo = git.Repo.clone_from( 

36 metadata.info.url, str(path), multi_options=["-c core.autocrlf=false"] 

37 ) 

38 except git.GitCommandError as e: 

39 raise DppGitCloneError(metadata, str(path), e.command, e.status, e.stdout) 

40 

41 return repo 

42 

43 

44def _git_checkout( 

45 repo: git.Repo, checkout_dir: Path, defect: taxonomy.Defect 

46) -> git.Repo: 

47 """ 

48 Checkout branch to the given commit. 

49 """ 

50 if not checkout_dir.exists(): 

51 try: 

52 # Pass '-f' in case worktree directory could be registered but removed. 

53 repo.git.worktree("add", "-f", str(checkout_dir.resolve()), defect.hash) 

54 except git.GitCommandError: 

55 raise DppGitWorktreeError(repo, str(checkout_dir.resolve()), defect) 

56 

57 # git worktree list --porcelain will output 

58 # $ worktree path 

59 # $ ... 

60 porcelain_output = repo.git.worktree("list", "--porcelain").split("\n\n") 

61 dir_start_index = len("worktree ") + 1 

62 directory_names = [ 

63 Path(output.splitlines()[0][dir_start_index:]).name 

64 for output in porcelain_output 

65 ] 

66 if checkout_dir.name not in directory_names: 

67 # Not sure if this is reachable. 

68 raise DppGitCheckoutError(repo, str(checkout_dir), defect) 

69 

70 try: 

71 return git.Repo(checkout_dir) 

72 except git.exc.InvalidGitRepositoryError: 

73 raise DppGitCheckoutInvalidRepositoryError(repo, str(checkout_dir), defect) 

74 

75 

76def _git_am(repo: git.Repo, patches: List[str]): 

77 """ 

78 Apply patches to checkout branch. 

79 """ 

80 # Invoke command manually, because it seems like GitPython has a bug with updating submodules. 

81 if repo.submodules: 

82 try: 

83 repo.git.execute(["git", "submodule", "update", "--init"]) 

84 except git.exc.GitError as e: 

85 raise DppGitSubmoduleInitError(repo, e.command, e.status, e.stderr) 

86 

87 prev_hash = repo.git.rev_parse("--verify", "HEAD") 

88 patches = list(filter(None, patches)) 

89 if patches: 

90 message.info( 

91 __name__, f"{', '.join(os.path.basename(patch) for patch in patches)}" 

92 ) 

93 else: 

94 message.info(__name__, "no patches") 

95 

96 for patch in patches: 

97 try: 

98 repo.git.am(patch) 

99 except git.GitCommandError as e: 

100 raise DppGitApplyPatchError(repo, patch, e.command, e.status, e.stderr) 

101 

102 current_hash = repo.git.rev_parse("--verify", "HEAD") 

103 if prev_hash == current_hash: 

104 raise DppGitPatchNotAppliedError(repo, patch) 

105 prev_hash = current_hash 

106 

107 

108class CheckoutCommand(Command): 

109 """ 

110 Checkout command which handles VCS commands based on taxonomy information. 

111 """ 

112 

113 _ERROR_MESSAGES: Dict[Type[DppGitError], str] = { 

114 DppGitCloneError: "git-clone failed", 

115 DppGitWorktreeError: "git-worktree failed", 

116 DppGitCheckoutInvalidRepositoryError: "git-checkout failed (not a git repository)", 

117 DppGitCheckoutError: "git-checkout failed", 

118 DppGitSubmoduleInitError: "git-submodule failed", 

119 DppGitApplyPatchError: "git-am failed", 

120 DppGitPatchNotAppliedError: "git-am patch could not be applied", 

121 } 

122 

123 def __init__(self): 

124 super().__init__() 

125 # TODO: write argparse description in detail 

126 self.parser = create_common_vcs_parser() 

127 self.parser.usage = "bugcpp.py checkout PROJECT INDEX [-b|--buggy] [-t|--target WORKSPACE] [-s|--source-only] [-v|--verbose]" 

128 self.parser.description = dedent( 

129 """\ 

130 Checkout defect taxonomy. 

131 """ 

132 ) 

133 

134 def __call__(self, argv: List[str]): 

135 """ 

136 Clone a repository into the given directory or checkout to a specific commit on the host machine. 

137 It does not perform action inside a container unlike the other commands. 

138 It utilizes git-worktree rather than cleaning up the current directory and checking out. 

139 It not only makes hoping around commits more quickly, but also reduces overhead of writing and deleting files. 

140 

141 Parameters 

142 ---------- 

143 argv : List[str] 

144 Command line argument vector. 

145 

146 Returns 

147 ------- 

148 None 

149 """ 

150 args = self.parser.parse_args(argv) 

151 metadata = args.metadata 

152 metadata_base = args.metadata_base 

153 worktree = args.worktree 

154 # args.index is 1 based. 

155 defect = metadata.defects[args.index - 1] 

156 

157 try: 

158 message.info(__name__, f"git-clone '{metadata.name}'") 

159 message.stdout_progress( 

160 f"[{metadata.name}] cloning a new repository from '{metadata.info.url}'" 

161 ) 

162 repo = _git_clone(worktree.base / ".repo", metadata) 

163 

164 message.info(__name__, "git-checkout") 

165 message.stdout_progress(f"[{metadata.name}] checking out '{defect.hash}'") 

166 checkout_repo = _git_checkout(repo, worktree.host, defect) 

167 

168 current_hash = checkout_repo.git.rev_parse("--verify", "HEAD") 

169 if current_hash == defect.hash: 

170 message.info(__name__, "git-am") 

171 _git_am( 

172 checkout_repo, 

173 [ 

174 defect.common_patch, 

175 defect.split_patch, 

176 defect.buggy_patch if args.buggy else defect.fixed_patch, 

177 ], 

178 ) 

179 else: 

180 message.info(__name__, "git-am skipped") 

181 

182 # check if there are extra data 

183 path_to_extra = Path(metadata_base).joinpath( 

184 metadata.name, "extra", f"{args.index:04}" 

185 ) 

186 if path_to_extra.exists(): 

187 message.info( 

188 __name__, 

189 f"copying extra directory(f{path_to_extra}) to project root", 

190 ) 

191 for extra in Path.iterdir(path_to_extra): 

192 shutil.copytree( 

193 extra, Path(worktree.host, extra.stem), dirs_exist_ok=True 

194 ) 

195 

196 message.info(__name__, f"creating '.defects4cpp.json' at {worktree.host}") 

197 # Write .defects4cpp.json in the directory. 

198 Project.write_config(worktree) 

199 if not args.source_only: 

200 try: 

201 docker = Docker(metadata.dockerfile, worktree, verbose=args.verbose) 

202 image = docker.image 

203 message.info(__name__, f" image: '{image}'") 

204 except Exception as e: 

205 message.stdout_error( 

206 f" An API Error occured.{os.linesep}" 

207 f" Find detailed message at {message.path}." 

208 ) 

209 

210 except DppGitError as e: 

211 message.error(__name__, str(e)) 

212 message.stdout_progress_error(f"[{metadata.name}] {str(e)}") 

213 sys.exit(1) 

214 

215 # pull docker image if it does not exist 

216 message.info(__name__, "pulling docker image") 

217 dockerfile = metadata.dockerfile 

218 tag = Path(dockerfile).parent.name 

219 self._container_name: str = f"{tag}-dpp" 

220 self._tag = f"hschoe/defects4cpp-ubuntu:{tag}" 

221 

222 message.stdout_progress(f"[{metadata.name}] done") 

223 

224 @property 

225 def group(self) -> str: 

226 return "v1" 

227 

228 @property 

229 def help(self) -> str: 

230 return "Get a specific defect snapshot"