1""" 

2Manage commands associated with docker SDK module. 

3 

4Do not use docker SDK directly, instead use Docker class. 

5""" 

6import os 

7import sys 

8from pathlib import Path 

9from textwrap import dedent 

10from typing import Dict, Optional, cast 

11 

12import docker.errors 

13from config import config 

14from docker import DockerClient 

15from docker.models.containers import Container, ExecResult 

16from docker.models.images import Image 

17from errors import DppError 

18from errors.docker import (DppDockerBuildClientError, DppDockerBuildError, DppDockerBuildServerError, 

19 DppDockerNoClientError) 

20from message import message 

21from processor.core.data import Worktree 

22 

23 

24def _cast_image(image) -> Image: 

25 """ 

26 Helper function to get a correct type 

27 """ 

28 return cast(Image, image) 

29 

30 

31def _cast_container(container) -> Container: 

32 """ 

33 Helper function to get a correct type 

34 """ 

35 return cast(Container, container) 

36 

37 

38def _try_build_image(client, tag, path, verbose) -> Image: 

39 try: 

40 image, stream = client.images.build(rm=True, tag=tag, path=path, pull=True) 

41 if verbose: 

42 for chunk in stream: 

43 if "stream" in chunk: 

44 for line in chunk["stream"].splitlines(): 

45 print(line) 

46 return image 

47 except docker.errors.BuildError as e: 

48 raise DppDockerBuildError(e.msg) 

49 except docker.errors.APIError as e: 

50 if e.is_client_error(): 

51 raise DppDockerBuildClientError(e.explanation) 

52 else: 

53 raise DppDockerBuildServerError(e.explanation) 

54 

55 

56def _build_image(client, tag, path, verbose) -> Image: 

57 """ 

58 Helper function to get a correct type 

59 """ 

60 try: 

61 return _try_build_image(client, tag, path, verbose) 

62 except DppError as e: 

63 message.stdout_progress_error(e) 

64 sys.exit(1) 

65 

66 

67class _Client: 

68 def __get__(self, instance, owner) -> DockerClient: 

69 if getattr(owner, "_client", None) is None: 

70 try: 

71 setattr(owner, "_client", docker.from_env()) 

72 except docker.errors.DockerException: 

73 message.error(__name__, "no response from docker-daemon") 

74 raise DppDockerNoClientError() 

75 return getattr(owner, "_client") 

76 

77 

78class Docker: 

79 """ 

80 Provide docker SDK methods along with context manager. 

81 It is recommended to use this via `with` statement. 

82 

83 Examples 

84 -------- 

85 >>> with Docker("/path/to/Dockerfile", my_worktree) as docker: 

86 ... docker.send("echo 'Hello, world!'") 

87 

88 Notes 

89 ----- 

90 Host machine must be running docker daemon in background. 

91 """ 

92 

93 client = _Client() 

94 """Docker SDK client.""" 

95 

96 def __init__( 

97 self, 

98 dockerfile: str, 

99 worktree: Worktree, 

100 environ: Optional[Dict[str, str]] = None, 

101 rebuild_image=False, 

102 user=None, 

103 uid_of_user: Optional[str] = None, 

104 verbose=True, 

105 ): 

106 self._dockerfile = dockerfile 

107 # Assume that the parent directory has the same name as the target. 

108 tag = Path(dockerfile).parent.name 

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

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

111 self._volume: Dict[str, Dict] = { 

112 str(worktree.host.resolve()): { 

113 "bind": str(worktree.container), 

114 "mode": "rw", 

115 } 

116 } 

117 self._working_dir: str = str(worktree.container) 

118 self._environ = environ 

119 self._rebuild_image = rebuild_image 

120 self._image: Optional[Image] = None 

121 self._container: Optional[Container] = None 

122 self._user = user 

123 self._uid_of_dpp_docker_user = uid_of_user 

124 self._verbose = verbose 

125 

126 @property 

127 def image(self) -> Image: 

128 """Docker SDK Image.""" 

129 if not self._image: 

130 message.stdout_progress_detail(f" Image: {self._tag}") 

131 try: 

132 # SET DEFECTS4CPP_TEST_TAXONOMY to 1 for taxonomy testing 

133 if os.environ.get("DEFECTS4CPP_TEST_TAXONOMY", "0") not in ["YES", "1"]: 

134 self.client.images.pull(self._tag) 

135 self._image = _cast_image(self.client.images.get(self._tag)) 

136 except docker.errors.ImageNotFound: 

137 message.info( 

138 f"ImageNotFound {self.client}, {self._tag}, {str(Path(self._dockerfile).parent)}" 

139 ) 

140 self._image = _build_image( 

141 self.client, 

142 self._tag, 

143 str(Path(self._dockerfile).parent), 

144 self._verbose, 

145 ) 

146 except docker.errors.APIError as api_error: 

147 message.stdout_error( 

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

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

150 ) 

151 message.error(__name__, f"APIError {api_error}") 

152 self._image = _build_image( 

153 self.client, 

154 self._tag, 

155 str(Path(self._dockerfile).parent), 

156 self._verbose, 

157 ) 

158 

159 if self._uid_of_dpp_docker_user is not None: 

160 # set uid of user(defects4cpp) to self._uid_of_dpp_docker_user after checking if uid need to be changed 

161 _container = _cast_container( 

162 self.client.containers.run( 

163 self._image, 

164 detach=True, 

165 stdin_open=True, 

166 environment=self._environ, 

167 name=self._container_name, 

168 ) 

169 ) 

170 _, output = _container.exec_run(f"id -u {config.DPP_DOCKER_USER}") 

171 uid = str(output, "utf-8").strip("\n") 

172 if self._uid_of_dpp_docker_user != uid: 

173 message.stdout_progress_detail( 

174 f" Setting uid of {config.DPP_DOCKER_USER} " 

175 f"from {uid} " 

176 f"to {self._uid_of_dpp_docker_user}." 

177 ) 

178 _, output = _container.exec_run( 

179 f"usermod -u {self._uid_of_dpp_docker_user} {config.DPP_DOCKER_USER}" 

180 ) 

181 repository, tag = tuple(self._tag.split(":")) 

182 self._image = _cast_image( 

183 _container.commit(repository=repository, tag=tag) 

184 ) 

185 _container.stop() 

186 _container.remove() 

187 return self._image 

188 

189 def __enter__(self): 

190 k = list(self._volume)[0] 

191 message.info( 

192 __name__, 

193 dedent( 

194 """container.__enter__ ({}) 

195 - from {} 

196 - to {} 

197 - mode {}""".format( 

198 self._container_name, 

199 DppError.print_path(k), 

200 self._volume[k]["bind"], 

201 self._volume[k]["mode"], 

202 ) 

203 ), 

204 ) 

205 message.stdout_progress_detail("Starting container") 

206 self._container = _cast_container( 

207 self.client.containers.run( 

208 self.image, 

209 auto_remove=True, 

210 command="/bin/sh", 

211 detach=True, 

212 environment=self._environ, 

213 name=self._container_name, 

214 stdin_open=True, 

215 user=config.DPP_DOCKER_USER if not self._user else self._user, 

216 volumes=self._volume, 

217 working_dir=self._working_dir, 

218 ) 

219 ) 

220 return self 

221 

222 def __exit__(self, type, value, traceback): 

223 message.info(__name__, f"container.__exit__ ({self._container_name})") 

224 message.stdout_progress_detail("Closing container") 

225 self._container.stop() 

226 

227 def send(self, command: str, stream=True, **kwargs) -> ExecResult: 

228 """ 

229 Send a single line command to the running container. 

230 """ 

231 return self._container.exec_run(command, stream=stream, **kwargs)