Coverage for src/features/pipelines.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from argparse import Namespace 

5 

6# Components 

7from ..jobs.runner import Runner 

8from ..models.flags import Flags 

9from ..models.pipelines import Pipeline 

10from ..package.bundle import Bundle 

11from ..prints.histories import PipelineHistory 

12from ..system.platform import Platform 

13from ..types.lists import Lists 

14 

15# PipelinesFeature class, pylint: disable=too-few-public-methods 

16class PipelinesFeature: 

17 

18 # Members 

19 __pipeline: Pipeline 

20 __options: Namespace 

21 

22 # Constructor 

23 def __init__( 

24 self, 

25 pipeline: Pipeline, 

26 options: Namespace, 

27 ) -> None: 

28 

29 # Prepare pipeline 

30 self.__pipeline = pipeline 

31 

32 # Prepare options 

33 self.__options = options 

34 

35 # Filter 

36 def __filter( 

37 self, 

38 job: str, 

39 ) -> bool: 

40 

41 # Filter pipeline jobs list, pylint: disable=duplicate-code 

42 if not self.__options.pipeline and not Lists.match( 

43 self.__options.names, 

44 job, 

45 ignore_case=self.__options.ignore_case, 

46 no_regex=self.__options.no_regex, 

47 ): 

48 return False 

49 

50 # Filter pipeline stages list 

51 if self.__options.pipeline and self.__options.names and not Lists.match( 

52 self.__options.names, 

53 self.__pipeline.jobs[job].stage, 

54 ignore_case=self.__options.ignore_case, 

55 no_regex=self.__options.no_regex, 

56 ): 

57 return False 

58 

59 # Filter manual pipeline jobs 

60 job_manual = self.__pipeline.jobs[job].when == 'manual' 

61 if job_manual and not Flags.check_bool( 

62 option=self.__options.manual, 

63 flag=Bundle.FLAG_MANUAL, 

64 variables=self.__pipeline.jobs[job].variables, 

65 default=False, 

66 ) and not Lists.match( 

67 self.__options.names, 

68 job, 

69 ignore_case=self.__options.ignore_case, 

70 no_regex=self.__options.no_regex, 

71 ): 

72 return False 

73 

74 # Filter disabled pipeline jobs 

75 if self.__pipeline.jobs[job].options.disabled: 

76 return False 

77 

78 # Result 

79 return True 

80 

81 # Launch 

82 def launch(self) -> bool: 

83 

84 # Variables 

85 notify: bool = False 

86 pipeline_history = PipelineHistory() 

87 result = None 

88 

89 # Prepare interacted flag 

90 pipeline_history.interacted = False 

91 

92 # Run selected pipeline jobs 

93 for job in self.__pipeline.jobs: 

94 

95 # Filter pipeline jobs 

96 if not self.__filter(job): 

97 continue 

98 

99 # Raise initial result 

100 if result is None: 

101 result = True 

102 

103 # Raise interacted flag 

104 if Flags.check_bool( 

105 option=self.__options.bash, 

106 flag=Bundle.FLAG_BASH, 

107 variables=self.__pipeline.jobs[job].variables, 

108 default=False, 

109 ) or Flags.check_bool( 

110 option=self.__options.debug, 

111 flag=Bundle.FLAG_DEBUG, 

112 variables=self.__pipeline.jobs[job].variables, 

113 default=False, 

114 ): 

115 pipeline_history.interacted = True 

116 

117 # Raise notify flag 

118 if not notify and Flags.check_bool( 

119 option=self.__options.notify, 

120 flag=Bundle.FLAG_NOTIFY, 

121 variables=self.__pipeline.jobs[job].variables, 

122 default=False, 

123 ): 

124 notify = True 

125 

126 # Run job 

127 expected = result 

128 result = Runner(options=self.__options).run( 

129 self.__pipeline.jobs[job], 

130 result, 

131 pipeline_history, 

132 ) 

133 

134 # Retry job if allowed 

135 attempt: int = 0 

136 if expected and not result and self.__pipeline.jobs[job].retry > 0: 

137 while not result and attempt < self.__pipeline.jobs[job].retry: 

138 attempt += 1 

139 result = Runner(options=self.__options).run( 

140 self.__pipeline.jobs[job], 

141 expected, 

142 pipeline_history, 

143 ) 

144 

145 # Update pipeline history 

146 pipeline_history.result = bool(result) 

147 

148 # Non quiet jobs 

149 if not pipeline_history.jobs_quiet and not self.__options.scripts: 

150 

151 # Pipeline jobs footer 

152 if pipeline_history.jobs_count > 1: 

153 

154 # Output pipeline history 

155 pipeline_history.print() 

156 

157 # Notify pipeline history 

158 if not pipeline_history.interacted and notify: 

159 pipeline_history.notify() 

160 

161 # Footer 

162 print(' ') 

163 Platform.flush() 

164 

165 # Result 

166 return bool(result)