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

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from enum import Enum 

5from os import environ 

6from sys import exit as sys_exit 

7from typing import Dict, List 

8 

9# Components 

10from ..models.images import Entrypoint 

11from ..package.bundle import Bundle 

12from ..prints.colors import Colors 

13from ..system.platform import Platform 

14from ..types.volumes import Volumes 

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

16from .docker import DockerEngine 

17from .podman import PodmanEngine 

18 

19# Backend enumeration 

20class Backend(Enum): 

21 DOCKER = 1 

22 PODMAN = 2 

23 UNKNOWN = 3 

24 

25# Names enumeration, pylint: disable=too-few-public-methods 

26class Names: 

27 

28 # Constants 

29 AUTO: str = 'auto' 

30 DOCKER: str = 'docker' 

31 PODMAN: str = 'podman' 

32 

33 # Defaults 

34 DEFAULTS: List[str] = [ 

35 DOCKER, 

36 PODMAN, 

37 ] 

38 

39 # Getter 

40 @staticmethod 

41 def get(override: str) -> List[str]: 

42 

43 # Adapt override 

44 override = override.lower() if override else '' 

45 

46 # Handle engine overrides 

47 if override: 

48 auto: bool = False 

49 names: List[str] = [] 

50 overrides: List[str] = override.split(',') 

51 for item in overrides: 

52 if item: 

53 if Names.AUTO == item: 

54 auto = True 

55 else: 

56 names += [name for name in Names.DEFAULTS if name == item] 

57 if auto or override[-1] == ',': 

58 names = names + Names.DEFAULTS 

59 names = list(dict.fromkeys(names)) 

60 

61 # Use engine defaults 

62 else: 

63 names = Names.DEFAULTS 

64 

65 # Result 

66 return names 

67 

68# Privileged default 

69def privileged_default() -> bool: 

70 if Platform.IS_WINDOWS: 

71 return False # pragma: windows cover 

72 return True 

73 

74# Supported engines 

75def supported() -> List[str]: 

76 return [Names.AUTO] + Names.DEFAULTS 

77 

78# Engine class 

79class Engine: 

80 

81 # Members 

82 __engine: BaseEngine 

83 __name: str = '' 

84 

85 # Constructor 

86 def __init__( 

87 self, 

88 engines_names: str, 

89 ) -> None: 

90 

91 # Variables 

92 engine: BaseEngine | None = None 

93 

94 # Acquire engine names 

95 names: List[str] = Names.get(engines_names) 

96 

97 # Iterate through names 

98 for name in names: 

99 

100 # Detect Docker engine 

101 if name == Names.DOCKER: 

102 try: # pragma: docker cover 

103 engine = DockerEngine() 

104 self.__name = Names.DOCKER 

105 break 

106 except (KeyboardInterrupt, ModuleNotFoundError, PermissionError): 

107 engine = None 

108 

109 # Detect Podman engine 

110 elif name == Names.PODMAN: 

111 try: # pragma: podman cover 

112 engine = PodmanEngine() 

113 self.__name = Names.PODMAN 

114 break 

115 except (KeyboardInterrupt, ModuleNotFoundError, PermissionError): 

116 engine = None 

117 

118 # Ignore pre-commit hook without engine 

119 if not engine and 'PRE_COMMIT' in environ: 

120 print(f' {Colors.GREEN}{Bundle.NAME}: {Colors.RED}ERROR:' 

121 f' {Colors.BOLD}Unknown or unsupported container engine...' 

122 f' {Colors.CYAN}README: {Bundle.REPOSITORY}#supported-container-engines' 

123 f' {Colors.YELLOW_LIGHT}(Ignored in \'pre-commit\' context)' 

124 f'{Colors.RESET}') 

125 print(' ') 

126 sys_exit(0) 

127 

128 # Unknown engine fallback 

129 if not engine: 

130 raise NotImplementedError( 

131 'Unknown or unsupported container engine...' 

132 f' README: {Bundle.REPOSITORY}#supported-container-engines') 

133 

134 # Store engine 

135 self.__engine = engine 

136 

137 # Command exec 

138 def cmd_exec(self) -> str: 

139 return self.__engine.cmd_exec() 

140 

141 # Container 

142 @property 

143 def container(self) -> ContainerName: 

144 return self.__engine.container 

145 

146 # Exec 

147 def exec(self, commands: Commands) -> _ExecResult: # pragma: no cover 

148 return self.__engine.exec(commands) 

149 

150 # Get 

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

152 self.__engine.get(image) 

153 

154 # Logs 

155 def logs(self) -> _LogsResult: 

156 return self.__engine.logs() 

157 

158 # Name 

159 @property 

160 def name(self) -> str: 

161 return self.__name 

162 

163 # Pull 

164 def pull( 

165 self, 

166 image: str, 

167 force: bool = False, 

168 ) -> None: 

169 self.__engine.pull(image, force=force) 

170 

171 # Remove 

172 def remove(self) -> None: 

173 self.__engine.remove() 

174 

175 # Remove image 

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

177 self.__engine.rmi(image) 

178 

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

180 def run( 

181 self, 

182 image: str, 

183 commands: Commands, 

184 entrypoint: Entrypoint, 

185 variables: Dict[str, str], 

186 network: str, 

187 option_privileged: bool, 

188 option_sockets: bool, 

189 services: bool, 

190 volumes: Volumes, 

191 directory: str, 

192 temp_folder: str, 

193 ) -> None: 

194 self.__engine.run( 

195 image=image, 

196 commands=commands, 

197 entrypoint=entrypoint, 

198 variables=variables, 

199 network=network, 

200 option_privileged=option_privileged, 

201 option_sockets=option_sockets, 

202 services=services, 

203 volumes=volumes, 

204 directory=directory, 

205 temp_folder=temp_folder, 

206 ) 

207 

208 # Stop 

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

210 self.__engine.stop(timeout) 

211 

212 # Supports 

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

214 return self.__engine.supports(binary) 

215 

216 # Wait 

217 def wait(self) -> bool: 

218 return self.__engine.wait()