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

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from __future__ import annotations 

5from os import environ 

6from subprocess import CompletedProcess, DEVNULL, PIPE, Popen, run 

7from typing import Dict, Iterator, List, Optional 

8 

9# Components 

10from ..models.images import Entrypoint 

11from ..package.bundle import Bundle 

12from ..system.platform import Platform 

13from ..types.volumes import Volumes 

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

15 

16# Podman engine class 

17class PodmanEngine(BaseEngine): # pragma: podman cover 

18 

19 # Members 

20 __binary: str = 'podman' 

21 __container: Optional[str] 

22 

23 # Constructor 

24 def __init__(self) -> None: 

25 

26 # Prepare container 

27 self.__container = None 

28 

29 # Configure binary 

30 if environ.get(Bundle.ENV_PODMAN_BINARY_PATH, ''): 

31 self.__binary = environ[Bundle.ENV_PODMAN_BINARY_PATH] 

32 

33 # Check engine support 

34 try: 

35 result = self.__exec([ 

36 'system', 

37 'info', 

38 ], True) 

39 if result.returncode != 0: 

40 raise ModuleNotFoundError() 

41 except FileNotFoundError: 

42 raise ModuleNotFoundError() from None 

43 

44 # Internal execution 

45 def __exec( 

46 self, 

47 arguments: List[str], 

48 quiet: bool = False, 

49 ) -> 'CompletedProcess[bytes]': 

50 

51 # Execute quiet command 

52 if quiet: 

53 return run( 

54 [self.__binary] + arguments, 

55 check=False, 

56 stdout=DEVNULL, 

57 stderr=DEVNULL, 

58 ) 

59 

60 # Execute standard command 

61 return run( 

62 [self.__binary] + arguments, 

63 check=False, 

64 stdout=PIPE, 

65 stderr=PIPE, 

66 ) 

67 

68 # Internal watcher 

69 def __watch(self, arguments: List[str]) -> Iterator[bytes]: 

70 

71 # Watch command outputs, pylint: disable=consider-using-with 

72 stdout = Popen( 

73 [self.__binary] + arguments, 

74 stdout=PIPE, 

75 ).stdout 

76 if not stdout: # pragma: no cover 

77 return iter() # type: ignore[call-overload,no-any-return] 

78 return iter(stdout.readline, b'') 

79 

80 # Command exec 

81 def cmd_exec(self) -> str: 

82 

83 # Result 

84 if Platform.IS_USER_SUDO: 

85 return f'sudo {self.__binary} exec -it' 

86 return f'{self.__binary} exec -it' 

87 

88 # Container 

89 @property 

90 def container(self) -> ContainerName: 

91 

92 # Result 

93 assert self.__container is not None 

94 result = self.__exec([ 

95 'inspect', 

96 '--type', 

97 'container', 

98 '--format', 

99 '{{.Name}}', 

100 self.__container, 

101 ]) 

102 return result.stdout.strip().decode('utf-8') if result.returncode == 0 else '' 

103 

104 # Exec 

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

106 

107 # Execute command in container 

108 assert self.__container is not None 

109 result = self.__exec([ 

110 'exec', 

111 self.__container, 

112 ] + commands) 

113 return _ExecResult( 

114 exit_code=result.returncode, 

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

116 ) 

117 

118 # Get 

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

120 

121 # Validate image exists 

122 result = self.__exec([ 

123 'inspect', 

124 '--type', 

125 'image', 

126 '--format', 

127 'exists', 

128 image, 

129 ], True) 

130 

131 # Pull missing image 

132 if result.returncode != 0: 

133 self.pull(image) 

134 

135 # Logs 

136 def logs(self) -> _LogsResult: 

137 

138 # Return logs stream 

139 assert self.__container is not None 

140 return self.__watch([ 

141 'logs', 

142 '--follow', 

143 self.__container, 

144 ]) 

145 

146 # Pull 

147 def pull( 

148 self, 

149 image: str, 

150 force: bool = False, 

151 ) -> None: 

152 

153 # Header 

154 print(f'Pulling from {image}') 

155 Platform.flush() 

156 

157 # Force image removal 

158 if force: 

159 self.rmi(image) 

160 

161 # Pull image with logs stream 

162 result = self.__exec([ 

163 'pull', 

164 image, 

165 ]) 

166 

167 # Layer completion logs 

168 if result.returncode == 0: 

169 result = self.__exec([ 

170 'inspect', 

171 '--type', 

172 'image', 

173 '--format', 

174 '{{.Id}}', 

175 image, 

176 ]) 

177 print(f"Digest: {result.stdout.strip().decode('utf-8')}") 

178 print(f'Status: Image is up to date for {image}') 

179 else: 

180 print(f'Status: Image not found for {image}') 

181 Platform.flush() 

182 raise FileNotFoundError(result.stderr.decode('utf-8').replace('\\n', '\n')) 

183 

184 # Footer 

185 print(' ') 

186 Platform.flush() 

187 

188 # Remove 

189 def remove(self) -> None: 

190 

191 # Remove container 

192 if self.__container: 

193 self.__exec([ 

194 'rm', 

195 '--force', 

196 self.__container, 

197 ]) 

198 self.__container = None 

199 

200 # Remove image 

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

202 

203 # Remove image 

204 result = self.__exec([ 

205 'inspect', 

206 '--type', 

207 'image', 

208 '--format', 

209 'exists', 

210 image, 

211 ], True) 

212 if result.returncode == 0: 

213 self.__exec([ 

214 'rmi', 

215 image, 

216 ]) 

217 

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

219 def run( 

220 self, 

221 image: str, 

222 commands: Commands, 

223 entrypoint: Entrypoint, 

224 variables: Dict[str, str], 

225 network: str, 

226 option_privileged: bool, 

227 option_sockets: bool, 

228 services: bool, 

229 volumes: Volumes, 

230 directory: str, 

231 temp_folder: str, 

232 ) -> None: 

233 

234 # Variables 

235 args_commands: Commands = [] 

236 args_entrypoint: List[str] = [] 

237 args_env: List[str] = [] 

238 args_run: List[str] = [] 

239 args_volumes: List[str] = [] 

240 

241 # Adapt command 

242 if commands: 

243 args_commands += commands 

244 

245 # Adapt entrypoint 

246 if isinstance(entrypoint, list): 

247 if len(entrypoint) > 1: 

248 args_commands = [' '.join(args_commands)] 

249 args_commands[0:0] = entrypoint[1:] 

250 entrypoint = entrypoint[0] 

251 if isinstance(entrypoint, str): 

252 args_entrypoint = ['--entrypoint', entrypoint] 

253 

254 # Adapt mounts 

255 if volumes: 

256 for volume in volumes.get().values(): 

257 args_volumes += ['--volume', f'{volume.stringify()}'] 

258 

259 # Adapt variables 

260 for variable in variables: 

261 args_env.extend(['--env', f'{variable}={variables[variable]}']) 

262 

263 # Prepare arguments 

264 args_run += ['create'] 

265 args_run += args_entrypoint 

266 args_run += args_env 

267 args_run += ['--tty'] 

268 args_run += args_volumes 

269 if network: 

270 args_run += ['--network', network] 

271 if option_privileged: 

272 args_run += ['--privileged'] 

273 args_run += ['--security-opt', 'label=disable'] 

274 args_run += ['--workdir', directory] 

275 args_run += [image] 

276 args_run += args_commands 

277 

278 # Create container image 

279 result = self.__exec(args_run) 

280 if result.returncode == 0: 

281 self.__container = result.stdout.strip().decode('utf-8') 

282 

283 # Handle creation failures 

284 else: 

285 raise NotImplementedError(result.stderr.decode('utf-8').replace('\\n', '\n')) 

286 

287 # Start container 

288 result = self.__exec([ 

289 'start', 

290 self.__container, 

291 ]) 

292 

293 # Handle start failures 

294 if result.returncode != 0: # pragma: no cover 

295 raise RuntimeError(result.stderr.decode('utf-8').replace('\\n', '\n')) 

296 

297 # Stop 

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

299 

300 # Stop container 

301 assert self.__container is not None 

302 self.__exec([ 

303 'stop', 

304 '--time', 

305 str(timeout), 

306 self.__container, 

307 ]) 

308 

309 # Supports 

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

311 

312 # Validate binary support 

313 result = self.exec([ 

314 'whereis', 

315 binary, 

316 ]) 

317 

318 # Result 

319 return result.exit_code == 0 

320 

321 # Wait 

322 def wait(self) -> bool: 

323 

324 # Wait container 

325 assert self.__container is not None 

326 result = self.__exec([ 

327 'wait', 

328 self.__container, 

329 ]) 

330 

331 # Result 

332 try: 

333 return int(result.stdout.strip()) == 0 if result.returncode == 0 else False 

334 except ValueError: 

335 return False