Coverage for tests/wms_test_utils.py: 94%

1# This file is part of ctrl_bps. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <https://www.gnu.org/licenses/>. 

27import dataclasses 

28import logging 

29 

30from lsst.ctrl.bps.wms_service import BaseWmsService, WmsJobReport, WmsRunReport, WmsStates 

31 

32_LOG = logging.getLogger(__name__) 

33 

34 

35TEST_REPORT = WmsRunReport( 

36 wms_id="1.0", 

37 global_wms_id="foo#1.0", 

38 path="/path/to/run", 

39 label="label", 

40 run="run", 

41 site="SITE1", 

42 project="dev", 

43 campaign="testing", 

44 payload="test", 

45 operator="tester", 

46 run_summary="foo:1", 

47 state=WmsStates.SUCCEEDED, 

48 jobs=[WmsJobReport(wms_id="1.0", name="", label="foo", state=WmsStates.SUCCEEDED)], 

49 total_number_jobs=1, 

50 job_state_counts={state: 1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates}, 

51 job_summary={ 

52 "foo": {state: 1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates}, 

53 }, 

54) 

55 

56 

57class WmsServiceSuccess(BaseWmsService): 

58 """WMS service class with working ping and get_status.""" 

59 

60 def ping(self, pass_thru): 

61 _LOG.info(f"Success {pass_thru}") 

62 return 0, "" 

63 

64 def report( 

65 self, 

66 wms_workflow_id=None, 

67 user=None, 

68 hist=0, 

69 pass_thru=None, 

70 is_global=False, 

71 return_exit_codes=False, 

72 ): 

73 report = dataclasses.replace(TEST_REPORT) 

74 return [report], [] 

75 

76 def get_status( 

77 self, 

78 wms_workflow_id, 

79 hist=0, 

80 pass_thru=None, 

81 is_global=False, 

82 ): 

83 return WmsStates.SUCCEEDED, "" 

84 

85 

86class WmsServiceFailure(BaseWmsService): 

87 """WMS service class with non-functional ping and 

88 get_status for failed run. 

89 """ 

90 

91 def ping(self, pass_thru): 

92 _LOG.warning("service failure") 

93 return 64, "Couldn't contact service X" 

94 

95 def report( 

96 self, 

97 wms_workflow_id=None, 

98 user=None, 

99 hist=0, 

100 pass_thru=None, 

101 is_global=False, 

102 return_exit_codes=False, 

103 ): 

104 report = WmsRunReport() 

105 return [report], [] 

106 

107 def get_status( 

108 self, 

109 wms_workflow_id, 

110 hist=0, 

111 pass_thru=None, 

112 is_global=False, 

113 ): 

114 return WmsStates.FAILED, "Dummy error message." 

115 

116 

117class WmsServicePassThru(BaseWmsService): 

118 """WMS service class with pass through ping.""" 

119 

120 def ping(self, pass_thru): 

121 _LOG.info(pass_thru) 

122 return 0, pass_thru 

123 

124 

125class WmsServiceDefault(BaseWmsService): 

126 """WMS service class with default ping and get_status.""" 

127 

128 def ping(self, pass_thru): 

129 _LOG.info(f"DEFAULT {pass_thru}") 

130 return 0, "default" 

131 

132 def get_status( 

133 self, 

134 wms_workflow_id=None, 

135 hist=0, 

136 pass_thru=None, 

137 is_global=False, 

138 ): 

139 return WmsStates.RUNNING, "" 

140 

141 

142class WmsServiceFromCmdline(BaseWmsService): 

143 """WMS service class with its own default settings.""" 

144 

145 @property 

146 def defaults(self): 

147 return {"corge": "cmdline"} 

148 

149 @property 

150 def defaults_path(self): 

151 return "/wms/class/from/cmdline" 

152 

153 

154class WmsServiceFromConfig(BaseWmsService): 

155 """WMS service class with its own default settings.""" 

156 

157 @property 

158 def defaults(self): 

159 return {"corge": "config"} 

160 

161 @property 

162 def defaults_path(self): 

163 return "/wms/class/from/config" 

164 

165 

166class WmsServiceFromEnv(BaseWmsService): 

167 """WMS service class with its own default settings.""" 

168 

169 @property 

170 def defaults(self): 

171 return {"corge": "env"} 

172 

173 @property 

174 def defaults_path(self): 

175 return "/wms/class/from/env" 

176 

177 

178class WmsServiceFromDefaults(BaseWmsService): 

179 """WMS service class with its own default settings.""" 

180 

181 @property 

182 def defaults(self): 

183 return {"corge": "defaults"} 

184 

185 @property 

186 def defaults_path(self): 

187 return "/wms/class/from/defaults" 

188 

189 

190class WmsServiceInvalid: 

191 """WMS service class that is not a subclass of BaseWmsService."""