Coverage for tests / test_bpsconfig.py: 23%

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 os 

28import unittest 

29 

30import yaml 

31 

32from lsst.ctrl.bps import BPS_SEARCH_ORDER, BpsConfig 

33from lsst.daf.butler import Config 

34 

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

36 

37 

38class TestBpsConfigConstructor(unittest.TestCase): 

39 """Test BpsConfig construction.""" 

40 

41 def setUp(self): 

42 self.filename = os.path.join(TESTDIR, "data/config.yaml") 

43 with open(self.filename) as f: 

44 self.dictionary = yaml.safe_load(f) 

45 

46 def tearDown(self): 

47 pass 

48 

49 def testFromFilename(self): 

50 """Test initialization from a file, but without defaults.""" 

51 config = BpsConfig(self.filename) 

52 self.assertEqual(set(config), {"foo", "bar", "baz"} | set(BPS_SEARCH_ORDER)) 

53 self.assertEqual(set(config["foo"]), {"qux"}) 

54 self.assertEqual(set(config["bar"]), {"qux"}) 

55 self.assertEqual(set(config["baz"]), {"grault", "garply"}) 

56 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

57 

58 def testFromDict(self): 

59 """Test initialization from a dictionary.""" 

60 config = BpsConfig(self.dictionary) 

61 self.assertEqual(set(config), {"foo", "bar", "baz"} | set(BPS_SEARCH_ORDER)) 

62 self.assertEqual(set(config["foo"]), {"qux"}) 

63 self.assertEqual(set(config["bar"]), {"qux"}) 

64 self.assertEqual(set(config["baz"]), {"grault", "garply"}) 

65 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

66 

67 def testFromConfig(self): 

68 """Test initialization from other Config object.""" 

69 config_daf = Config(self.dictionary) 

70 config = BpsConfig(config_daf) 

71 self.assertEqual(set(config), {"foo", "bar", "baz"} | set(BPS_SEARCH_ORDER)) 

72 self.assertEqual(set(config["foo"]), {"qux"}) 

73 self.assertEqual(set(config["bar"]), {"qux"}) 

74 self.assertEqual(set(config["baz"]), {"grault", "garply"}) 

75 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

76 

77 def testFromBpsConfig(self): 

78 """Test initialization from other BpsConfig object.""" 

79 config_ref = BpsConfig(self.dictionary) 

80 config = BpsConfig(config_ref) 

81 self.assertEqual(set(config), set(config_ref)) 

82 self.assertEqual(config.search_order, config_ref.search_order) 

83 

84 def testDefaultsInclusion(self): 

85 """Test including defaults.""" 

86 config = BpsConfig(self.filename, defaults={"quux": 1}) 

87 self.assertEqual(set(config), {"foo", "bar", "baz", "quux"} | set(BPS_SEARCH_ORDER)) 

88 self.assertEqual(config["quux"], 1) 

89 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

90 

91 def testDefaultsOverride(self): 

92 """Test overriding defaults.""" 

93 config = BpsConfig(self.filename, defaults={"foo": {"qux": 2}, "quux": 1}) 

94 self.assertEqual(set(config), {"foo", "bar", "baz", "quux"} | set(BPS_SEARCH_ORDER)) 

95 self.assertEqual(config["quux"], 1) 

96 self.assertEqual(config["foo"], BpsConfig({"qux": 1}, search_order=[])) 

97 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

98 

99 def testWmsFromCmdline(self): 

100 """Test initialization when WMS class specified at the cmdline.""" 

101 with unittest.mock.patch.dict( 

102 os.environ, {"BPS_WMS_SERVICE_CLASS": "wms_test_utils.WmsServiceFromEnv"} 

103 ): 

104 filename = os.path.join(TESTDIR, "data/config_with_wms.yaml") 

105 config = BpsConfig( 

106 filename, 

107 defaults={"wmsServiceClass": "wms_test_utils.WmsServiceFromDefaults"}, 

108 wms_service_class_fqn="wms_test_utils.WmsServiceFromCmdline", 

109 ) 

110 self.assertEqual(set(config), {"corge", "wmsServiceClass"} | set(BPS_SEARCH_ORDER)) 

111 self.assertEqual(config["corge"], "cmdline") 

112 

113 def testWmsFromConfig(self): 

114 """Test initialization when WMS class specified in the config.""" 

115 with unittest.mock.patch.dict( 

116 os.environ, {"BPS_WMS_SERVICE_CLASS": "wms_test_utils.WmsServiceFromEnv"} 

117 ): 

118 filename = os.path.join(TESTDIR, "data/config_with_wms.yaml") 

119 config = BpsConfig( 

120 filename, 

121 defaults={"wmsServiceClass": "wms_test_utils.WmsServiceFromDefaults"}, 

122 wms_service_class_fqn=None, 

123 ) 

124 self.assertEqual(set(config), {"corge", "wmsServiceClass"} | set(BPS_SEARCH_ORDER)) 

125 self.assertEqual(config["corge"], "config") 

126 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

127 

128 def testWmsFromEnv(self): 

129 """Test initialization with WMS class in the env.""" 

130 with unittest.mock.patch.dict( 

131 os.environ, {"BPS_WMS_SERVICE_CLASS": "wms_test_utils.WmsServiceFromEnv"} 

132 ): 

133 config = BpsConfig( 

134 {}, 

135 defaults={"wmsServiceClass": "wms_test_utils.WmsServiceFromDefaults"}, 

136 wms_service_class_fqn=None, 

137 ) 

138 self.assertEqual(set(config), {"corge", "wmsServiceClass"} | set(BPS_SEARCH_ORDER)) 

139 self.assertEqual(config["corge"], "env") 

140 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

141 

142 def testWmsFromDefaults(self): 

143 """Test initialization with a default WMS class.""" 

144 config = BpsConfig({}, defaults={"wmsServiceClass": "wms_test_utils.WmsServiceFromDefaults"}) 

145 self.assertEqual(set(config), {"corge", "wmsServiceClass"} | set(BPS_SEARCH_ORDER)) 

146 self.assertEqual(config["corge"], "defaults") 

147 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

148 

149 def testDefaultsExclusion(self): 

150 config = BpsConfig( 

151 self.filename, defaults=None, wms_service_class_fqn="wms_test_utils.WmsServiceFromCmdline" 

152 ) 

153 self.assertEqual(set(config), {"foo", "bar", "baz"} | set(BPS_SEARCH_ORDER)) 

154 self.assertEqual(set(config["foo"]), {"qux"}) 

155 self.assertEqual(set(config["bar"]), {"qux"}) 

156 self.assertEqual(set(config["baz"]), {"grault", "garply"}) 

157 self.assertEqual(config.search_order, BPS_SEARCH_ORDER) 

158 

159 def testInvalidArg(self): 

160 """Test if exception is raised for an argument of unsupported type.""" 

161 sequence = ["wibble", "wobble", "wubble", "flob"] 

162 with self.assertRaises(ValueError): 

163 BpsConfig(sequence) 

164 

165 

166class TestBpsConfigGet(unittest.TestCase): 

167 """Test retrieval of items from BpsConfig.""" 

168 

169 def setUp(self): 

170 self.config = BpsConfig({"foo": "bar"}) 

171 

172 def tearDown(self): 

173 pass 

174 

175 def testKeyExistsNoDefault(self): 

176 """Test if the value is returned when the key is in the dictionary.""" 

177 self.assertEqual("bar", self.config.get("foo")) 

178 

179 def testKeyExistsDefaultProvided(self): 

180 """Test if the value is returned when the key is in the dictionary.""" 

181 self.assertEqual("bar", self.config.get("foo", "qux")) 

182 

183 def testKeyMissingNoDefault(self): 

184 """Test if the provided default is returned if the key is missing.""" 

185 self.assertEqual("", self.config.get("baz")) 

186 

187 def testKeyMissingDefaultProvided(self): 

188 """Test if the custom default is returned if the key is missing.""" 

189 self.assertEqual("qux", self.config.get("baz", "qux")) 

190 

191 

192class TestBpsConfigSearch(unittest.TestCase): 

193 """Test searching of BpsConfig.""" 

194 

195 def setUp(self): 

196 filename = os.path.join(TESTDIR, "data/config.yaml") 

197 self.config = BpsConfig(filename, search_order=["baz", "bar", "foo"], defaults={}) 

198 os.environ["GARPLY"] = "garply" 

199 

200 def tearDown(self): 

201 del os.environ["GARPLY"] 

202 

203 def testSectionSearchOrder(self): 

204 """Test if sections are searched in the prescribed order.""" 

205 key = "qux" 

206 found, value = self.config.search(key) 

207 self.assertEqual(found, True) 

208 self.assertEqual(value, 2) 

209 

210 def testCurrentValues(self): 

211 """Test if a current value overrides of the one in configuration.""" 

212 found, value = self.config.search("qux", opt={"curvals": {"qux": -3}}) 

213 self.assertEqual(found, True) 

214 self.assertEqual(value, -3) 

215 

216 def testSearchobjValues(self): 

217 """Test if a serachobj value overrides of the one in configuration.""" 

218 options = {"searchobj": {"qux": 4}} 

219 found, value = self.config.search("qux", opt=options) 

220 self.assertEqual(found, True) 

221 self.assertEqual(value, 4) 

222 

223 def testSubsectionSearch(self): 

224 options = {"curvals": {"curr_baz": "garply"}} 

225 found, value = self.config.search("qux", opt=options) 

226 self.assertEqual(found, True) 

227 self.assertEqual(value, 3) 

228 

229 def testDefault(self): 

230 """Test if a default value is properly set.""" 

231 found, value = self.config.search("plugh", opt={"default": 4}) 

232 self.assertEqual(found, True) 

233 self.assertEqual(value, 4) 

234 

235 def testReplaceVars(self): 

236 """Test replaceVars method.""" 

237 test_opt = {"default": 555} 

238 orig_str = "<ENV:GARPLY>/waldo/{qux:03}/{notthere}" 

239 value = self.config.replace_vars(orig_str, opt=test_opt) 

240 self.assertEqual(value, "<ENV:GARPLY>/waldo/002/") 

241 self.assertEqual(test_opt["default"], 555) 

242 

243 def testReplaceVarsSkipNames(self): 

244 test_opt = {"default": 555, "skipNames": ["qgraphFile", "butlerConfig"]} 

245 orig_str = "<ENV:GARPLY>/waldo/{qux:03} {qgraphFile} {butlerConfig}" 

246 value = self.config.replace_vars(orig_str, opt=test_opt) 

247 self.assertEqual(value, "<ENV:GARPLY>/waldo/002 {qgraphFile} {butlerConfig}") 

248 self.assertEqual(test_opt["default"], 555) 

249 

250 def testVariables(self): 

251 """Test combinations of expandEnvVars, replaceEnvVars, 

252 and replaceVars. 

253 """ 

254 test_opt = {"expandEnvVars": False, "replaceEnvVars": False, "replaceVars": False} 

255 found, value = self.config.search("grault", opt=test_opt) 

256 self.assertEqual(found, True) 

257 self.assertEqual(value, "${GARPLY}/waldo/{qux:03}") 

258 

259 test_opt = {"expandEnvVars": False, "replaceEnvVars": False, "replaceVars": True} 

260 found, value = self.config.search("grault", opt=test_opt) 

261 self.assertEqual(found, True) 

262 self.assertEqual(value, "${GARPLY}/waldo/002") 

263 

264 test_opt = {"expandEnvVars": False, "replaceEnvVars": True, "replaceVars": False} 

265 found, value = self.config.search("grault", opt=test_opt) 

266 self.assertEqual(found, True) 

267 self.assertEqual(value, "<ENV:GARPLY>/waldo/{qux:03}") 

268 

269 test_opt = {"expandEnvVars": False, "replaceEnvVars": True, "replaceVars": True} 

270 found, value = self.config.search("grault", opt=test_opt) 

271 self.assertEqual(found, True) 

272 self.assertEqual(value, "<ENV:GARPLY>/waldo/002") 

273 

274 test_opt = {"expandEnvVars": True, "replaceEnvVars": False, "replaceVars": False} 

275 found, value = self.config.search("grault", opt=test_opt) 

276 self.assertEqual(found, True) 

277 self.assertEqual(value, "garply/waldo/{qux:03}") 

278 

279 test_opt = {"expandEnvVars": True, "replaceEnvVars": False, "replaceVars": True} 

280 found, value = self.config.search("grault", opt=test_opt) 

281 self.assertEqual(found, True) 

282 self.assertEqual(value, "garply/waldo/002") 

283 

284 test_opt = {"expandEnvVars": True, "replaceEnvVars": True, "replaceVars": False} 

285 found, value = self.config.search("grault", opt=test_opt) 

286 self.assertEqual(found, True) 

287 self.assertEqual(value, "garply/waldo/{qux:03}") 

288 

289 test_opt = {"expandEnvVars": True, "replaceEnvVars": True, "replaceVars": True} 

290 found, value = self.config.search("grault", opt=test_opt) 

291 self.assertEqual(found, True) 

292 self.assertEqual(value, "garply/waldo/002") 

293 

294 def testRequired(self): 

295 """Test if exception is raised if a required setting is missing.""" 

296 with self.assertRaises(KeyError): 

297 self.config.search("fred", opt={"required": True}) 

298 

299 

300class TestBpsConfigGenerateConfig(unittest.TestCase): 

301 """Test BpsConfig.generate_config and bpsEval methods.""" 

302 

303 def setUp(self): 

304 # Just to shorten string length in tests 

305 self.test_prefix = "lsst.ctrl.bps.tests.config_test_utils" 

306 

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

308 self.config = BpsConfig(filename, BPS_SEARCH_ORDER, defaults={}) 

309 

310 def testUnparsableValue(self): 

311 config = BpsConfig( 

312 { 

313 "p1": 3, 

314 "p3": 16, 

315 }, 

316 BPS_SEARCH_ORDER, 

317 ) 

318 # invalid function name 

319 config["bpsGenerateConfig"] = "not a valid name({p1}, param3={p3})" 

320 with self.assertRaisesRegex(ValueError, "Unparsable bpsGenerateConfig value='not a valid"): 

321 config.generate_config() 

322 

323 def testMissingParen(self): 

324 config = BpsConfig( 

325 { 

326 "p1": 3, 

327 "p3": 16, 

328 }, 

329 BPS_SEARCH_ORDER, 

330 ) 

331 # invalid function name 

332 config["bpsGenerateConfig"] = self.test_prefix + ".generate_config_1(1, param3=2" 

333 with self.assertRaisesRegex(ValueError, "Unparsable bpsGenerateConfig value='"): 

334 config.generate_config() 

335 

336 def testBadFunctionName(self): 

337 config = BpsConfig( 

338 { 

339 "p1": 3, 

340 "p3": 16, 

341 }, 

342 BPS_SEARCH_ORDER, 

343 ) 

344 # invalid function name 

345 config["bpsGenerateConfig"] = self.test_prefix + ".notthere({p1}, param3={p3})" 

346 with self.assertRaisesRegex(ImportError, "notthere"): 

347 config.generate_config() 

348 

349 def testBadModuleName(self): 

350 config = BpsConfig( 

351 { 

352 "p1": 3, 

353 "p3": 16, 

354 }, 

355 BPS_SEARCH_ORDER, 

356 ) 

357 # invalid module name 

358 config["bpsGenerateConfig"] = "lsst.ctrl.bps.notthere.generate_config(1, 2)" 

359 with self.assertRaisesRegex(ImportError, "notthere"): 

360 config.generate_config() 

361 

362 def testBadParamName(self): 

363 config = BpsConfig( 

364 {"p1": 3, "p3": 16, "bpsGenerateConfig": self.test_prefix + ".generate_config_1(1, param5=2)"}, 

365 BPS_SEARCH_ORDER, 

366 ) 

367 with self.assertRaisesRegex(TypeError, "unexpected keyword argument 'param5'"): 

368 config.generate_config() 

369 

370 def testExtraParam(self): 

371 config = BpsConfig( 

372 { 

373 "p1": 3, 

374 "p3": 16, 

375 "bpsGenerateConfig": self.test_prefix + ".generate_config_1({p1}, 2, {p3}, 4)", 

376 }, 

377 BPS_SEARCH_ORDER, 

378 ) 

379 with self.assertRaisesRegex(TypeError, "positional arguments"): 

380 config.generate_config() 

381 

382 def testWithSearchOrder(self): 

383 # Check that bpsGenerateConfig is replaced in search sections 

384 # (e.g., pipetask) And when replacing vars in subsections 

385 # config ordering is used. 

386 # Ditto for finalJob (which isn't a search section). 

387 # Checking all in single function to ensure doesn't quit early. 

388 self.config.generate_config() 

389 

390 filename = os.path.join(TESTDIR, "data/initialize_config_truth.yaml") 

391 truth = BpsConfig(filename, BPS_SEARCH_ORDER, defaults={}) 

392 

393 self.assertEqual(self.config, truth) 

394 

395 def testBpsEval(self): 

396 """Test replacing bpsEval when need to import module.""" 

397 test_opt = { 

398 "expandEnvVars": True, 

399 "replaceEnvVars": True, 

400 "replaceVars": True, 

401 "curvals": {"curr_pipetask": "ptask1"}, 

402 } 

403 found, value = self.config.search("genval1", opt=test_opt) 

404 self.assertEqual(found, True) 

405 self.assertEqual(value, "-c val1:0.1 -c val2:3") 

406 

407 def testBpsEvalBuiltin(self): 

408 """Test replacing bpsEval with builtin function.""" 

409 test_opt = { 

410 "expandEnvVars": True, 

411 "replaceEnvVars": True, 

412 "replaceVars": True, 

413 "curvals": {"curr_pipetask": "ptask1"}, 

414 } 

415 found, value = self.config.search("genval2", opt=test_opt) 

416 self.assertEqual(found, True) 

417 self.assertEqual(value, "-c val1:32") 

418 

419 def testBpsEvalInvalid(self): 

420 """Test reporting not replacing bpsEval.""" 

421 test_opt = { 

422 "expandEnvVars": True, 

423 "replaceEnvVars": True, 

424 "replaceVars": True, 

425 "curvals": {"curr_pipetask": "ptask1"}, 

426 } 

427 

428 self.config["badkey1"] = "badval1 bpsEval('sum([1,2]') blah" 

429 with self.assertRaisesRegex(ValueError, "Unparsable bpsEval in 'badval1 bpsEval"): 

430 _ = self.config.search("badkey1", opt=test_opt) 

431 

432 

433if __name__ == "__main__": 

434 unittest.main()