Coverage for tests / test_initialize.py: 32%

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/>. 

27 

28"""Unit tests for initialize.py""" 

29 

30import logging 

31import os 

32import tempfile 

33import unittest 

34from pathlib import Path 

35 

36from lsst.ctrl.bps import BpsConfig 

37from lsst.ctrl.bps.initialize import ( 

38 init_submission, 

39 out_collection_validator, 

40 output_run_validator, 

41 submit_path_validator, 

42) 

43 

44TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

45 

46 

47class TestOutCollectionValidator(unittest.TestCase): 

48 """Checks that 'outCollection' is *not* specified in config. 

49 

50 Assumes BpsConfig tests cover whether it finds values. 

51 """ 

52 

53 def testSuccess(self): 

54 config = BpsConfig({}, defaults={}) 

55 out_collection_validator(config) 

56 

57 def testFailure(self): 

58 config = BpsConfig({"outCollection": "dummy_collection"}, defaults={}) 

59 with self.assertRaisesRegex( 

60 KeyError, "'outCollection' is deprecated. Replace all references to it with 'outputRun'" 

61 ): 

62 out_collection_validator(config) 

63 

64 

65class TestOutputRunValidator(unittest.TestCase): 

66 """Checks that 'outputRun' is specified in config. 

67 

68 Assumes BpsConfig tests cover whether it finds values. 

69 """ 

70 

71 def testSuccess(self): 

72 config = BpsConfig({"outputRun": "dummy_run_value"}, defaults={}) 

73 output_run_validator(config) 

74 

75 def testFailure(self): 

76 config = BpsConfig({}, defaults={}) 

77 with self.assertRaisesRegex(KeyError, "Must specify the output run collection using 'outputRun'"): 

78 output_run_validator(config) 

79 

80 

81class TestSubmitPathValidator(unittest.TestCase): 

82 """Check that 'submitPath' is specified in BPS config. 

83 

84 Assumes BpsConfig tests cover whether it finds values. 

85 """ 

86 

87 def testSuccess(self): 

88 config = BpsConfig({"submitPath": "submit/dummy/path"}, defaults={}) 

89 submit_path_validator(config) 

90 

91 def testFailure(self): 

92 config = BpsConfig({}, defaults={}) 

93 with self.assertRaisesRegex( 

94 KeyError, "Must specify the submit-side run directory using 'submitPath'" 

95 ): 

96 submit_path_validator(config) 

97 

98 

99class TestInitSubmission(unittest.TestCase): 

100 """Check init_submission_function.""" 

101 

102 def setUp(self): 

103 self.temp_dir = tempfile.TemporaryDirectory() 

104 

105 def tearDown(self): 

106 self.temp_dir.cleanup() 

107 

108 def testBasicSuccess(self): 

109 filename = os.path.join(TESTDIR, "data/initialize_config.yaml") 

110 validators = [submit_path_validator, output_run_validator, out_collection_validator] 

111 config = init_submission( 

112 filename, 

113 validators=validators, 

114 compute_site="local", 

115 runWmsSubmissionChecks=False, 

116 wms_service="wms_test_utils.WmsServiceSuccess", 

117 tempDir=self.temp_dir.name, 

118 multVal=["a", "b", "c"], 

119 ) 

120 self.assertIn(".bps_defined.timestamp", config) 

121 self.assertIn(".bps_defined.operator", config) 

122 self.assertIn(".bps_defined.uniqProcName", config) 

123 self.assertIn(".bps_defined.submitPath", config) 

124 self.assertEqual(config[".bps_cmdline.computeSite"], "local") 

125 self.assertEqual(config[".bps_cmdline.multVal"], "a,b,c") 

126 uniq_proc_name = config[".bps_defined.uniqProcName"] 

127 submit_path = Path(config[".bps_defined.submitPath"]).resolve() 

128 files = [f.name for f in submit_path.iterdir() if f.is_file()] 

129 self.assertIn("initialize_config.yaml", files) 

130 self.assertIn(f"{uniq_proc_name}_config.yaml", files) 

131 self.assertIn(f"{uniq_proc_name}.env.info.yaml", files) 

132 self.assertIn(f"{uniq_proc_name}.pkg.info.yaml", files) 

133 

134 # generate_config tested elsewhere so just 

135 # check a couple values that shows it ran. 

136 self.assertEqual(config[".genall_1"], "/repo/test") 

137 self.assertEqual(config[".pipetask.ptask1.p3"], 32) 

138 self.assertEqual(config[".finalJob.gencfg_4"], 9) 

139 

140 @unittest.mock.patch("lsst.ctrl.bps.initialize.BPS_DEFAULTS", {}) 

141 def testMissingWmsServiceClass(self): 

142 filename = os.path.join(TESTDIR, "data/initialize_config.yaml") 

143 

144 with self.assertRaisesRegex(KeyError, "Missing wmsServiceClass in bps config. Aborting."): 

145 _ = init_submission(filename, runWmsSubmissionChecks=True, tempDir=self.temp_dir.name) 

146 

147 @unittest.mock.patch("lsst.ctrl.bps.initialize.BPS_DEFAULTS", {}) 

148 def testSubmissionChecksNotImplemented(self): 

149 filename = os.path.join(TESTDIR, "data/initialize_config.yaml") 

150 

151 with self.assertLogs("lsst.ctrl.bps.initialize", level=logging.DEBUG) as cm: 

152 config = init_submission( 

153 filename, 

154 runWmsSubmissionChecks=True, 

155 tempDir=self.temp_dir.name, 

156 wms_service="wms_test_utils.WmsServiceSuccess", 

157 ) 

158 output = " ".join(cm.output) 

159 self.assertIn("run_submission_checks is not implemented in wms_test_utils.WmsServiceSuccess.", output) 

160 self.assertIn(".bps_defined.timestamp", config) 

161 uniq_proc_name = config[".bps_defined.uniqProcName"] 

162 submit_path = Path(config[".bps_defined.submitPath"]).resolve() 

163 files = [f.name for f in submit_path.iterdir() if f.is_file()] 

164 self.assertIn("initialize_config.yaml", files) 

165 self.assertIn(f"{uniq_proc_name}_config.yaml", files) 

166 self.assertIn(f"{uniq_proc_name}.env.info.yaml", files) 

167 self.assertIn(f"{uniq_proc_name}.pkg.info.yaml", files) 

168 

169 @unittest.mock.patch( 

170 "lsst.ctrl.bps.initialize.BPS_DEFAULTS", {"operator": "testuser", "uniqProcName": "uniqval"} 

171 ) 

172 def testAlreadySet(self): 

173 """Test if operator and uniqProcName already set.""" 

174 filename = os.path.join(TESTDIR, "data/initialize_config.yaml") 

175 

176 config = init_submission(filename, runWmsSubmissionChecks=False, tempDir=self.temp_dir.name) 

177 self.assertEqual(config["operator"], "testuser") 

178 self.assertNotIn(".bps_defined.operator", config) 

179 self.assertEqual(config["uniqProcName"], "uniqval") 

180 self.assertNotIn(".bps_defined.uniqProcName", config)