Coverage for src/engines/docker.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from os import environ 

5from typing import Any, cast, Dict, Optional, TYPE_CHECKING 

6 

7# Modules libraries 

8if TYPE_CHECKING: # pragma: no cover, pylint: disable=wrong-import-position 

9 from docker import DockerClient 

10 from docker.models.containers import Container 

11else: 

12 DockerClient = Any 

13 Container = Any 

14 

15# Components 

16from ..jobs.outputs import Outputs 

17from ..models.images import Entrypoint 

18from ..system.platform import Platform 

19from ..types.volumes import Volumes 

20from .base import _ExecResult, _LogsResult, BaseEngine, Commands, ContainerName 

21 

22# Docker engine class 

23class DockerEngine(BaseEngine): # pragma: docker cover 

24 

25 # Constants 

26 ENV_DOCKER_CERT_PATH: str = 'DOCKER_CERT_PATH' 

27 ENV_DOCKER_HOST: str = 'DOCKER_HOST' 

28 ENV_DOCKER_TLS_VERIFY: str = 'DOCKER_TLS_VERIFY' 

29 

30 # Members 

31 __client: DockerClient 

32 __container: Optional[Container] 

33 

34 # Constructor 

35 def __init__(self) -> None: 

36 

37 # Modules libraries, pylint: disable=import-outside-toplevel 

38 from docker import from_env 

39 from docker.errors import DockerException 

40 

41 # Prepare container 

42 self.__container = None 

43 

44 # Engine client 

45 try: 

46 self.__client = from_env() 

47 self.__client.ping() 

48 except DockerException: 

49 raise ModuleNotFoundError() from None 

50 

51 # Sockets, pylint: disable=no-self-use 

52 def __sockets( 

53 self, 

54 variables: Dict[str, str], 

55 volumes: Volumes, 

56 ) -> None: 

57 

58 # Variables 

59 docker_host = '' 

60 

61 # Detect TLS configurations 

62 if DockerEngine.ENV_DOCKER_TLS_VERIFY in environ: 

63 variables[DockerEngine.ENV_DOCKER_TLS_VERIFY] = environ[ 

64 DockerEngine.ENV_DOCKER_TLS_VERIFY] 

65 

66 # Detect certificates configurations 

67 if environ.get(DockerEngine.ENV_DOCKER_CERT_PATH, ''): # pragma: no local cover 

68 variables[DockerEngine.ENV_DOCKER_CERT_PATH] = '/certs' 

69 volumes.add( 

70 environ[DockerEngine.ENV_DOCKER_CERT_PATH], 

71 '/certs', 

72 'ro', 

73 True, 

74 ) 

75 

76 # Detect host configurations 

77 if environ.get(DockerEngine.ENV_DOCKER_HOST, ''): 

78 docker_host = environ[DockerEngine.ENV_DOCKER_HOST] 

79 

80 # Network Docker socket 

81 if docker_host[0:7] == 'http://' or docker_host[0:6] == 'tcp://': 

82 variables[DockerEngine.ENV_DOCKER_HOST] = docker_host # pragma: no local cover 

83 

84 # Local Docker socket 

85 elif docker_host[0:7] == 'unix://': # pragma: no cover 

86 volumes.add( 

87 docker_host[7:], 

88 docker_host[7:], 

89 'rw', 

90 True, 

91 ) 

92 

93 # Default Docker socket 

94 elif not docker_host: # pragma: no cover 

95 

96 # Add socket volume 

97 if Platform.IS_LINUX or Platform.IS_WINDOWS or Platform.IS_EXPERIMENTAL: 

98 volumes.add( 

99 '/var/run/docker.sock', 

100 '/var/run/docker.sock', 

101 'rw', 

102 True, 

103 ) 

104 

105 # Unavailable feature 

106 else: 

107 Outputs.warning('The Docker sockets feature is not available...') 

108 

109 # Unknown feature 

110 else: # pragma: no cover 

111 Outputs.warning( 

112 f'The {DockerEngine.ENV_DOCKER_HOST} = {docker_host}' \ 

113 ' configuration is not supported yet...' 

114 ) 

115 

116 # Command exec, pylint: disable=no-self-use 

117 def cmd_exec(self) -> str: 

118 

119 # Result 

120 return 'docker exec -it' 

121 

122 # Container 

123 @property 

124 def container(self) -> ContainerName: 

125 

126 # Result 

127 assert self.__container is not None 

128 return cast(str, self.__container.name) 

129 

130 # Exec 

131 def exec(self, commands: Commands) -> _ExecResult: 

132 

133 # Execute command in container 

134 assert self.__container is not None 

135 result = self.__container.exec_run(' '.join(commands)) 

136 return _ExecResult( 

137 exit_code=result.exit_code, 

138 output=result.output.strip().decode('utf-8'), 

139 ) 

140 

141 # Get 

142 def get(self, image: str) -> None: 

143 

144 # Modules libraries, pylint: disable=import-outside-toplevel 

145 from docker.errors import ImageNotFound 

146 

147 # Validate image exists 

148 try: 

149 self.__client.images.get(image) 

150 

151 # Pull missing image 

152 except ImageNotFound: 

153 self.pull(image) 

154 

155 # Logs 

156 def logs(self) -> _LogsResult: 

157 

158 # Return logs stream 

159 assert self.__container is not None 

160 return self.__container.logs(stream=True) # type: ignore[no-any-return] 

161 

162 # Pull 

163 def pull( 

164 self, 

165 image: str, 

166 force: bool = False, 

167 ) -> None: 

168 

169 # Force image removal 

170 if force: 

171 self.rmi(image) 

172 

173 # Pull image with logs stream 

174 for data in self.__client.api.pull(image, stream=True, decode=True): 

175 

176 # Layer progress logs (deprecated) 

177 if 'progress' in data: # pragma: no cover 

178 if Platform.IS_TTY_STDOUT: 

179 print(f"\r\x1b[K{data['id']}: {data['status']} {data['progress']}", 

180 end='') 

181 Platform.flush() 

182 

183 # Layer event logs 

184 elif 'progressDetail' in data: 

185 if Platform.IS_TTY_STDOUT: 

186 print(f"\r\x1b[K{data['id']}: {data['status']}", end='') 

187 Platform.flush() 

188 

189 # Layer completion logs 

190 elif 'id' in data: 

191 print(f"\r\x1b[K{data['id']}: {data['status']}") 

192 Platform.flush() 

193 

194 # Layer error logs (deprecated) 

195 elif 'error' in data: # pragma: no cover 

196 print(f"\r\x1b[K{data['error']}") 

197 Platform.flush() 

198 

199 # Layer status logs 

200 elif 'status' in data: 

201 print(f"\r\x1b[K{data['status']}") 

202 Platform.flush() 

203 

204 # Unsupported data logs 

205 else: # pragma: no cover 

206 print(f"\r\x1b[K{data}") 

207 Platform.flush() 

208 

209 # Footer 

210 print(' ') 

211 Platform.flush() 

212 

213 # Remove 

214 def remove(self) -> None: 

215 

216 # Remove container 

217 if self.__container: 

218 self.__container.remove(force=True) 

219 self.__container = None 

220 

221 # Remove image 

222 def rmi(self, image: str) -> None: 

223 

224 # Modules libraries, pylint: disable=import-outside-toplevel 

225 from docker.errors import ImageNotFound 

226 

227 # Remove image 

228 try: 

229 self.__client.api.remove_image(image) 

230 print(f'Untagged: {image}') 

231 except ImageNotFound: 

232 print(f'No such image: {image}') 

233 

234 # Footer 

235 print(' ') 

236 Platform.flush() 

237 

238 # Run, pylint: disable=too-many-arguments,too-many-positional-arguments 

239 def run( 

240 self, 

241 image: str, 

242 commands: Commands, 

243 entrypoint: Entrypoint, 

244 variables: Dict[str, str], 

245 network: str, 

246 option_privileged: bool, 

247 option_sockets: bool, 

248 services: bool, 

249 volumes: Volumes, 

250 directory: str, 

251 temp_folder: str, 

252 ) -> None: 

253 

254 # Append sockets mounts 

255 if option_sockets: 

256 self.__sockets(variables, volumes) 

257 

258 # Run container image 

259 self.__container = self.__client.containers.run( 

260 image=image, 

261 command=commands, 

262 detach=True, 

263 entrypoint=entrypoint, 

264 environment=variables, 

265 network_mode=network if network else 'bridge', 

266 privileged=option_privileged, 

267 remove=False, 

268 stdout=True, 

269 stderr=True, 

270 stream=True, 

271 volumes=volumes.flatten(), 

272 working_dir=directory, 

273 ) 

274 

275 # Stop 

276 def stop(self, timeout: int) -> None: 

277 

278 # Stop container 

279 assert self.__container is not None 

280 self.__container.stop(timeout=timeout) 

281 

282 # Supports 

283 def supports(self, binary: str) -> bool: 

284 

285 # Modules libraries, pylint: disable=import-outside-toplevel 

286 from docker.errors import APIError 

287 

288 # Variables 

289 exit_code: int = 1 

290 

291 # Validate binary support 

292 try: 

293 exit_code = self.exec([ 

294 'whereis', 

295 f'{binary}', 

296 ]).exit_code 

297 except APIError: # pragma: no cover 

298 pass 

299 

300 # Result 

301 return exit_code == 0 

302 

303 # Wait 

304 def wait(self) -> bool: 

305 

306 # Wait container 

307 assert self.__container is not None 

308 result = self.__container.wait() 

309 

310 # Result 

311 return cast(int, result['StatusCode']) == 0