Coverage for src/cli/entrypoint.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from argparse import Namespace 

5from enum import Enum 

6from os import environ 

7from pathlib import Path 

8from subprocess import check_output, DEVNULL, Popen 

9from sys import argv 

10from typing import Optional 

11 

12# Components 

13from ..engines.engine import supported as engine_supported 

14from ..features.configurations import ConfigurationsFeature 

15from ..features.images import ImagesFeature 

16from ..features.menus import MenusFeature 

17from ..features.pipelines import PipelinesFeature 

18from ..models.pipelines import Pipeline 

19from ..package.bundle import Bundle 

20from ..package.settings import Settings 

21from ..parsers.parsers import Parsers 

22from ..prints.colors import Colors 

23from ..system.platform import Platform 

24from ..types.files import Files 

25 

26# Entrypoint class, pylint: disable=too-few-public-methods,too-many-statements 

27class Entrypoint: 

28 

29 # Enumerations 

30 Result = Enum('Result', [ 

31 'SUCCESS', 

32 'FINALIZE', 

33 'ERROR', 

34 'CRITICAL', 

35 ]) 

36 

37 # CLI, pylint: disable=too-many-branches,too-many-locals,too-many-return-statements 

38 @staticmethod 

39 def cli( 

40 options: Namespace, 

41 settings: Settings, 

42 ) -> Result: 

43 

44 # Variables 

45 hint: str = '' 

46 interactive: bool = Platform.IS_TTY_STDIN and Platform.IS_TTY_STDOUT 

47 result: bool = False 

48 

49 # Prepare configuration 

50 if Path(options.configuration).is_dir(): 

51 options.configuration = Path(options.configuration) / Bundle.CONFIGURATION 

52 

53 # Prepare engine 

54 if options.engine is None and environ.get(Bundle.FLAG_ENGINE, ''): 

55 options.engine = environ[Bundle.FLAG_ENGINE] 

56 options.engine_default = True 

57 elif options.engine is not None: 

58 environ[Bundle.FLAG_ENGINE] = options.engine 

59 options.engine_default = False 

60 elif settings.has(group='engines', key='engine'): 

61 options.engine = settings.get(group='engines', key='engine') 

62 options.engine_default = True 

63 else: 

64 options.engine = ','.join(engine_supported()) 

65 options.engine_default = True 

66 settings.set(group='engines', key='engine', value=options.engine) 

67 

68 # Prepare no_console 

69 if not options.no_console: 

70 if settings.has('runner', 'no_console'): 

71 options.no_console = settings.get_bool('runner', 'no_console') 

72 else: 

73 options.no_console = False 

74 settings.set_bool('runner', 'no_console', options.no_console) 

75 

76 # Prepare no_git_safeties 

77 if not options.no_git_safeties: 

78 if settings.has('runner', 'no_git_safeties'): 

79 options.no_git_safeties = settings.get_bool('runner', 'no_git_safeties') 

80 else: 

81 options.no_git_safeties = False 

82 settings.set_bool('runner', 'no_git_safeties', options.no_git_safeties) 

83 

84 # Prepare no_script_fail 

85 if not options.no_script_fail: 

86 if settings.has('parsers', 'no_script_fail'): 

87 options.no_script_fail = settings.get_bool('parsers', 'no_script_fail') 

88 else: 

89 options.no_script_fail = False 

90 settings.set_bool('parsers', 'no_script_fail', options.no_script_fail) 

91 

92 # Prepare paths 

93 options.configuration = Path(options.configuration).resolve() 

94 options.path = options.configuration.parent 

95 

96 # Prepare tags 

97 if options.tags: 

98 options.tags = options.tags.split(',') 

99 else: 

100 options.tags = Bundle.ARGUMENT_TAGS_DEFAULT 

101 options.tags_default = True 

102 

103 # Read configuration 

104 pipeline: Optional[Pipeline] = Parsers(options).read() 

105 if not pipeline: 

106 return Entrypoint.Result.CRITICAL 

107 

108 # Header 

109 print(' ') 

110 Platform.flush() 

111 

112 # Dump configuration 

113 if options.dump: 

114 result = ConfigurationsFeature( 

115 pipeline=pipeline, 

116 options=options, 

117 ).dump() 

118 

119 # Pull jobs images 

120 elif options.pull: 

121 result = ImagesFeature( 

122 pipeline=pipeline, 

123 options=options, 

124 ).pull() 

125 

126 # Remove jobs images 

127 elif options.rmi: 

128 result = ImagesFeature( 

129 pipeline=pipeline, 

130 options=options, 

131 ).rmi() 

132 

133 # Select job 

134 elif options.list and interactive: 

135 options.manual = True 

136 options.no_regex = True 

137 result = MenusFeature( 

138 pipeline=pipeline, 

139 options=options, 

140 ).select() 

141 

142 # Select jobs 

143 elif options.select and interactive: 

144 options.no_regex = True 

145 result = MenusFeature( 

146 pipeline=pipeline, 

147 options=options, 

148 ).select() 

149 

150 # Select jobs with presets 

151 elif options.presets and interactive: 

152 result = MenusFeature( 

153 pipeline=pipeline, 

154 options=options, 

155 ).select() 

156 

157 # Launch pipeline or jobs 

158 elif options.pipeline or options.names: 

159 result = PipelinesFeature( 

160 pipeline=pipeline, 

161 options=options, 

162 ).launch() 

163 

164 # Select jobs 

165 elif interactive: 

166 options.no_regex = True 

167 result = MenusFeature( 

168 pipeline=pipeline, 

169 options=options, 

170 ).select() 

171 

172 # Launch all jobs 

173 elif options.all: 

174 options.pipeline = True 

175 result = PipelinesFeature( 

176 pipeline=pipeline, 

177 options=options, 

178 ).launch() 

179 

180 # Unsupported case 

181 else: 

182 

183 # Windows WinPTY compatibility 

184 if Platform.IS_WINDOWS and Bundle.ENV_WINPTY not in environ: # pragma: windows cover 

185 

186 # Prepare WinPTY variables 

187 hint = ' (on Windows, winpty is required)' 

188 winpty: Optional[str] = None 

189 if environ.get(Bundle.ENV_WINPTY_PATH, ''): 

190 winpty = environ[Bundle.ENV_WINPTY_PATH] 

191 

192 # Acquire WinPTY path 

193 try: 

194 if not winpty: 

195 winpty = str( 

196 check_output( 

197 args=['where', 'winpty.exe'], 

198 stderr=DEVNULL, 

199 ).strip()) 

200 except FileNotFoundError: # pragma: no cover 

201 pass 

202 else: 

203 

204 # Nested WinPTY launch 

205 _environ = environ.copy() 

206 _environ[Bundle.ENV_WINPTY] = 'true' 

207 try: 

208 with Popen( 

209 args=[winpty] + argv if winpty else argv, 

210 env=_environ, 

211 ) as process: 

212 process.wait() 

213 return Entrypoint.Result.CRITICAL 

214 except OSError: # pragma: no cover 

215 pass 

216 

217 # Unsupported interactive terminal 

218 print( 

219 f' {Colors.GREEN}{Bundle.NAME}: {Colors.RED}ERROR: ' \ 

220 f'{Colors.BOLD}Unsupported non-interactive context{hint}...{Colors.RESET}' 

221 ) 

222 print(' ') 

223 Platform.flush() 

224 

225 # Cleanup at exit 

226 Files.clean() 

227 

228 # Result 

229 return Entrypoint.Result.SUCCESS if result else Entrypoint.Result.ERROR