Coverage for tests / test_prepare_utils.py: 26%

1# This file is part of ctrl_bps_htcondor. 

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 prepare utility functions.""" 

29 

30import logging 

31import os 

32import unittest 

33 

34from lsst.ctrl.bps import ( 

35 BPS_DEFAULTS, 

36 BPS_SEARCH_ORDER, 

37 BpsConfig, 

38 GenericWorkflowExec, 

39 GenericWorkflowFile, 

40 GenericWorkflowJob, 

41) 

42from lsst.ctrl.bps.htcondor import prepare_utils 

43from lsst.ctrl.bps.tests.gw_test_utils import make_3_label_workflow, make_3_label_workflow_groups_sort 

44 

45logger = logging.getLogger("lsst.ctrl.bps.htcondor") 

46 

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

48 

49 

50class TranslateJobCmdsTestCase(unittest.TestCase): 

51 """Test _translate_job_cmds method.""" 

52 

53 def setUp(self): 

54 self.gw_exec = GenericWorkflowExec("test_exec", "/dummy/dir/pipetask") 

55 self.cached_vals = {"profile": {}, "bpsUseShared": True, "memoryLimit": 32768} 

56 

57 def testRetryUnlessNone(self): 

58 gwjob = GenericWorkflowJob("retryUnless", "label1", executable=self.gw_exec) 

59 gwjob.retry_unless_exit = None 

60 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

61 self.assertNotIn("retry_until", htc_commands) 

62 

63 def testRetryUnlessInt(self): 

64 gwjob = GenericWorkflowJob("retryUnlessInt", "label1", executable=self.gw_exec) 

65 gwjob.retry_unless_exit = 3 

66 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

67 self.assertEqual(int(htc_commands["retry_until"]), gwjob.retry_unless_exit) 

68 

69 def testRetryUnlessList(self): 

70 gwjob = GenericWorkflowJob("retryUnlessList", "label1", executable=self.gw_exec) 

71 gwjob.retry_unless_exit = [1, 2] 

72 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

73 self.assertEqual(htc_commands["retry_until"], "member(ExitCode, {1,2})") 

74 

75 def testRetryUnlessBad(self): 

76 gwjob = GenericWorkflowJob("retryUnlessBad", "label1", executable=self.gw_exec) 

77 gwjob.retry_unless_exit = "1,2,3" 

78 with self.assertRaises(ValueError) as cm: 

79 _ = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

80 self.assertIn("retryUnlessExit", str(cm.exception)) 

81 

82 def testEnvironmentBasic(self): 

83 gwjob = GenericWorkflowJob("jobEnvironment", "label1", executable=self.gw_exec) 

84 gwjob.environment = {"TEST_INT": 1, "TEST_STR": "TWO"} 

85 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

86 self.assertEqual(htc_commands["environment"], "TEST_INT=1 TEST_STR='TWO'") 

87 

88 def testEnvironmentSpaces(self): 

89 gwjob = GenericWorkflowJob("jobEnvironment", "label1", executable=self.gw_exec) 

90 gwjob.environment = {"TEST_SPACES": "spacey value"} 

91 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

92 self.assertEqual(htc_commands["environment"], "TEST_SPACES='spacey value'") 

93 

94 def testEnvironmentSingleQuotes(self): 

95 gwjob = GenericWorkflowJob("jobEnvironment", "label1", executable=self.gw_exec) 

96 gwjob.environment = {"TEST_SINGLE_QUOTES": "spacey 'quoted' value"} 

97 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

98 self.assertEqual(htc_commands["environment"], "TEST_SINGLE_QUOTES='spacey ''quoted'' value'") 

99 

100 def testEnvironmentDoubleQuotes(self): 

101 gwjob = GenericWorkflowJob("jobEnvironment", "label1", executable=self.gw_exec) 

102 gwjob.environment = {"TEST_DOUBLE_QUOTES": 'spacey "double" value'} 

103 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

104 self.assertEqual(htc_commands["environment"], """TEST_DOUBLE_QUOTES='spacey ""double"" value'""") 

105 

106 def testEnvironmentWithEnvVars(self): 

107 gwjob = GenericWorkflowJob("jobEnvironment", "label1", executable=self.gw_exec) 

108 gwjob.environment = {"TEST_ENV_VAR": "<ENV:CTRL_BPS_DIR>/tests"} 

109 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

110 self.assertEqual(htc_commands["environment"], "TEST_ENV_VAR='$ENV(CTRL_BPS_DIR)/tests'") 

111 

112 def testPeriodicRelease(self): 

113 gwjob = GenericWorkflowJob("periodicRelease", "label1", executable=self.gw_exec) 

114 gwjob.request_memory = 2048 

115 gwjob.memory_multiplier = 2 

116 gwjob.number_of_retries = 3 

117 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

118 release = ( 

119 "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " 

120 "(HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " 

121 "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " 

122 "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768" 

123 ) 

124 self.assertEqual(htc_commands["periodic_release"], release) 

125 

126 def testPeriodicRemoveNoRetries(self): 

127 gwjob = GenericWorkflowJob("periodicRelease", "label1", executable=self.gw_exec) 

128 gwjob.request_memory = 2048 

129 gwjob.memory_multiplier = 1 

130 gwjob.number_of_retries = 0 

131 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) 

132 remove = "JobStatus == 5 && (NumJobStarts > JobMaxRetries)" 

133 self.assertEqual(htc_commands["periodic_remove"], remove) 

134 self.assertEqual(htc_commands["max_retries"], 0) 

135 

136 

137class GroupToSubdagTestCase(unittest.TestCase): 

138 """Test _group_to_subdag function.""" 

139 

140 def testBlocking(self): 

141 gw = make_3_label_workflow_groups_sort("test1", True) 

142 gwjob = gw.get_job("group_order1_10001") 

143 config = BpsConfig( 

144 {}, 

145 search_order=BPS_SEARCH_ORDER, 

146 defaults=BPS_DEFAULTS, 

147 ) 

148 

149 htc_job = prepare_utils._group_to_subdag(config, gwjob, "the_prefix") 

150 self.assertEqual(len(htc_job.subdag), len(gwjob)) 

151 

152 

153class GatherSiteValuesTestCase(unittest.TestCase): 

154 """Test _gather_site_values function.""" 

155 

156 def testAllThere(self): 

157 config = BpsConfig( 

158 {}, 

159 search_order=BPS_SEARCH_ORDER, 

160 defaults=BPS_DEFAULTS, 

161 ) 

162 compute_site = "notThere" 

163 results = prepare_utils._gather_site_values(config, compute_site) 

164 self.assertEqual(results["memoryLimit"], BPS_DEFAULTS["memoryLimit"]) 

165 

166 def testNotSpecified(self): 

167 config = BpsConfig( 

168 {}, 

169 search_order=BPS_SEARCH_ORDER, 

170 defaults=BPS_DEFAULTS, 

171 wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", 

172 ) 

173 compute_site = "notThere" 

174 results = prepare_utils._gather_site_values(config, compute_site) 

175 self.assertEqual(results["memoryLimit"], BPS_DEFAULTS["memoryLimit"]) 

176 

177 

178class GatherLabelValuesTestCase(unittest.TestCase): 

179 """Test _gather_labels_values function.""" 

180 

181 def testClusterLabel(self): 

182 # Test cluster value overrides pipetask. 

183 label = "label1" 

184 config = BpsConfig( 

185 { 

186 "cluster": { 

187 "label1": { 

188 "releaseExpr": "cluster_val", 

189 "overwriteJobFiles": False, 

190 "profile": {"condor": {"prof_val1": 3}}, 

191 } 

192 }, 

193 "pipetask": {"label1": {"releaseExpr": "pipetask_val"}}, 

194 }, 

195 search_order=BPS_SEARCH_ORDER, 

196 defaults=BPS_DEFAULTS, 

197 wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", 

198 ) 

199 results = prepare_utils._gather_label_values(config, label) 

200 self.assertEqual( 

201 results, 

202 { 

203 "attrs": {}, 

204 "profile": {"prof_val1": 3}, 

205 "releaseExpr": "cluster_val", 

206 "overwriteJobFiles": False, 

207 }, 

208 ) 

209 

210 def testPipetaskLabel(self): 

211 label = "label1" 

212 config = BpsConfig( 

213 { 

214 "pipetask": { 

215 "label1": { 

216 "releaseExpr": "pipetask_val", 

217 "overwriteJobFiles": False, 

218 "profile": {"condor": {"prof_val1": 3}}, 

219 } 

220 } 

221 }, 

222 search_order=BPS_SEARCH_ORDER, 

223 defaults=BPS_DEFAULTS, 

224 wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", 

225 ) 

226 results = prepare_utils._gather_label_values(config, label) 

227 self.assertEqual( 

228 results, 

229 { 

230 "attrs": {}, 

231 "profile": {"prof_val1": 3}, 

232 "releaseExpr": "pipetask_val", 

233 "overwriteJobFiles": False, 

234 }, 

235 ) 

236 

237 def testNoSection(self): 

238 label = "notThere" 

239 config = BpsConfig( 

240 {}, 

241 search_order=BPS_SEARCH_ORDER, 

242 defaults=BPS_DEFAULTS, 

243 wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", 

244 ) 

245 results = prepare_utils._gather_label_values(config, label) 

246 self.assertEqual(results, {"attrs": {}, "profile": {}, "overwriteJobFiles": True}) 

247 

248 def testNoOverwriteSpecified(self): 

249 label = "notthere" 

250 config = BpsConfig( 

251 {}, 

252 search_order=BPS_SEARCH_ORDER, 

253 defaults={}, 

254 wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", 

255 ) 

256 results = prepare_utils._gather_label_values(config, label) 

257 self.assertEqual(results, {"attrs": {}, "profile": {}, "overwriteJobFiles": True}) 

258 

259 def testFinalJob(self): 

260 label = "finalJob" 

261 config = BpsConfig( 

262 {"finalJob": {"profile": {"condor": {"prof_val2": 6, "+attr_val1": 5}}}}, 

263 search_order=BPS_SEARCH_ORDER, 

264 defaults=BPS_DEFAULTS, 

265 wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", 

266 ) 

267 results = prepare_utils._gather_label_values(config, label) 

268 self.assertEqual( 

269 results, {"attrs": {"attr_val1": 5}, "profile": {"prof_val2": 6}, "overwriteJobFiles": False} 

270 ) 

271 

272 

273class CreateCheckJobTestCase(unittest.TestCase): 

274 """Test _create_check_job function.""" 

275 

276 def testSuccess(self): 

277 group_job_name = "group_order1_val1a" 

278 job_label = "order1" 

279 job = prepare_utils._create_check_job(group_job_name, job_label) 

280 self.assertIn(group_job_name, job.name) 

281 self.assertEqual(job.label, job_label) 

282 self.assertIn("check_group_status.sub", job.subfile) 

283 

284 

285class CreatePeriodicReleaseExprTestCase(unittest.TestCase): 

286 """Test _create_periodic_release_expr function.""" 

287 

288 def testNoReleaseExpr(self): 

289 results = prepare_utils._create_periodic_release_expr(2048, 1, 32768, "") 

290 self.assertEqual(results, "") 

291 

292 def testMultiplierNone(self): 

293 results = prepare_utils._create_periodic_release_expr(2048, None, 32768, "") 

294 self.assertEqual(results, "") 

295 

296 def testJustMemoryReleaseExpr(self): 

297 self.maxDiff = None # so test error shows entire strings 

298 results = prepare_utils._create_periodic_release_expr(2048, 2, 32768, "") 

299 truth = ( 

300 "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " 

301 "(HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " 

302 "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " 

303 "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768" 

304 ) 

305 self.assertEqual(results, truth) 

306 

307 def testJustUserReleaseExpr(self): 

308 results = prepare_utils._create_periodic_release_expr(2048, 1, 32768, "True") 

309 truth = "JobStatus == 5 && NumJobStarts <= JobMaxRetries && HoldReasonCode =!= 1 && True" 

310 self.assertEqual(results, truth) 

311 

312 def testJustUserReleaseExprMultiplierNone(self): 

313 results = prepare_utils._create_periodic_release_expr(2048, None, 32768, "True") 

314 truth = "JobStatus == 5 && NumJobStarts <= JobMaxRetries && HoldReasonCode =!= 1 && True" 

315 self.assertEqual(results, truth) 

316 

317 def testMemoryAndUserReleaseExpr(self): 

318 self.maxDiff = None # so test error shows entire strings 

319 results = prepare_utils._create_periodic_release_expr(2048, 2, 32768, "True") 

320 truth = ( 

321 "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " 

322 "((HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " 

323 "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " 

324 "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768 || " 

325 "HoldReasonCode =!= 1 && True)" 

326 ) 

327 self.assertEqual(results, truth) 

328 

329 

330class CreatePeriodicRemoveExprTestCase(unittest.TestCase): 

331 """Test _create_periodic_release_expr function.""" 

332 

333 def testBasicRemoveExpr(self): 

334 """Function assumes only called if max_retries >= 0.""" 

335 results = prepare_utils._create_periodic_remove_expr(2048, 1, 32768) 

336 truth = "JobStatus == 5 && (NumJobStarts > JobMaxRetries)" 

337 self.assertEqual(results, truth) 

338 

339 def testBasicRemoveExprMultiplierNone(self): 

340 """Function assumes only called if max_retries >= 0.""" 

341 results = prepare_utils._create_periodic_remove_expr(2048, None, 32768) 

342 truth = "JobStatus == 5 && (NumJobStarts > JobMaxRetries)" 

343 self.assertEqual(results, truth) 

344 

345 def testMemoryRemoveExpr(self): 

346 self.maxDiff = None # so test error shows entire strings 

347 results = prepare_utils._create_periodic_remove_expr(2048, 2, 32768) 

348 truth = ( 

349 "JobStatus == 5 && (NumJobStarts > JobMaxRetries || " 

350 "((HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " 

351 "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " 

352 "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) == 32768))" 

353 ) 

354 self.assertEqual(results, truth) 

355 

356 

357class HandleJobOutputsTestCase(unittest.TestCase): 

358 """Test _handle_job_outputs function.""" 

359 

360 def setUp(self): 

361 self.job_name = "test_job" 

362 self.out_prefix = "/test/prefix" 

363 

364 def tearDown(self): 

365 pass 

366 

367 def testNoOutputsSharedFilesystem(self): 

368 """Test with shared filesystem and no outputs.""" 

369 mock_workflow = unittest.mock.Mock() 

370 mock_workflow.get_job_outputs.return_value = [] 

371 

372 result = prepare_utils._handle_job_outputs(mock_workflow, self.job_name, True, self.out_prefix) 

373 

374 self.assertEqual(result, {"transfer_output_files": '""'}) 

375 

376 def testWithOutputsSharedFilesystem(self): 

377 """Test with shared filesystem and outputs present (still empty).""" 

378 mock_workflow = unittest.mock.Mock() 

379 mock_workflow.get_job_outputs.return_value = [ 

380 GenericWorkflowFile(name="output.txt", src_uri="/path/to/output.txt") 

381 ] 

382 

383 result = prepare_utils._handle_job_outputs(mock_workflow, self.job_name, True, self.out_prefix) 

384 

385 self.assertEqual(result, {"transfer_output_files": '""'}) 

386 

387 def testNoOutputsNoSharedFilesystem(self): 

388 """Test without shared filesystem and no outputs.""" 

389 mock_workflow = unittest.mock.Mock() 

390 mock_workflow.get_job_outputs.return_value = [] 

391 

392 result = prepare_utils._handle_job_outputs(mock_workflow, self.job_name, False, self.out_prefix) 

393 

394 self.assertEqual(result, {"transfer_output_files": '""'}) 

395 

396 def testWithAnOutputNoSharedFilesystem(self): 

397 """Test without shared filesystem and single output file.""" 

398 mock_workflow = unittest.mock.Mock() 

399 mock_workflow.get_job_outputs.return_value = [ 

400 GenericWorkflowFile(name="output.txt", src_uri="/path/to/output.txt") 

401 ] 

402 

403 result = prepare_utils._handle_job_outputs(mock_workflow, self.job_name, False, self.out_prefix) 

404 

405 expected = { 

406 "transfer_output_files": "output.txt", 

407 "transfer_output_remaps": '"output.txt=/path/to/output.txt"', 

408 } 

409 self.assertEqual(result, expected) 

410 

411 def testWithOutputsNoSharedFilesystem(self): 

412 """Test without shared filesystem and multiple output files.""" 

413 mock_workflow = unittest.mock.Mock() 

414 mock_workflow.get_job_outputs.return_value = [ 

415 GenericWorkflowFile(name="output1.txt", src_uri="/path/output1.txt"), 

416 GenericWorkflowFile(name="output2.txt", src_uri="/another/path/output2.txt"), 

417 ] 

418 

419 result = prepare_utils._handle_job_outputs(mock_workflow, self.job_name, False, self.out_prefix) 

420 

421 expected = { 

422 "transfer_output_files": "output1.txt,output2.txt", 

423 "transfer_output_remaps": '"output1.txt=/path/output1.txt;output2.txt=/another/path/output2.txt"', 

424 } 

425 self.assertEqual(result, expected) 

426 

427 @unittest.mock.patch("lsst.ctrl.bps.htcondor.prepare_utils._LOG") 

428 def testLogging(self, mock_log): 

429 mock_workflow = unittest.mock.Mock() 

430 mock_workflow.get_job_outputs.return_value = [ 

431 GenericWorkflowFile(name="output.txt", src_uri="/path/to/output.txt") 

432 ] 

433 

434 prepare_utils._handle_job_outputs(mock_workflow, self.job_name, False, self.out_prefix) 

435 

436 self.assertTrue(mock_log.debug.called) 

437 debug_calls = mock_log.debug.call_args_list 

438 self.assertTrue(any("src_uri=" in str(call) for call in debug_calls)) 

439 self.assertTrue(any("transfer_output_files=" in str(call) for call in debug_calls)) 

440 self.assertTrue(any("transfer_output_remaps=" in str(call) for call in debug_calls)) 

441 

442 

443class CreateJobTestCase(unittest.TestCase): 

444 """Test _create_job function.""" 

445 

446 def setUp(self): 

447 self.generic_workflow = make_3_label_workflow("test1", True) 

448 

449 def testNoOverwrite(self): 

450 template = "{label}/{tract}/{patch}/{band}/{subfilter}/{physical_filter}/{visit}/{exposure}" 

451 cached_values = { 

452 "bpsUseShared": True, 

453 "overwriteJobFiles": False, 

454 "memoryLimit": 491520, 

455 "profile": {}, 

456 "attrs": {}, 

457 } 

458 gwjob = self.generic_workflow.get_final() 

459 out_prefix = "submit" 

460 htc_job = prepare_utils._create_job(template, cached_values, self.generic_workflow, gwjob, out_prefix) 

461 self.assertEqual(htc_job.name, gwjob.name) 

462 self.assertEqual(htc_job.label, gwjob.label) 

463 self.assertIn("NumJobStarts", htc_job.cmds["output"]) 

464 self.assertIn("NumJobStarts", htc_job.cmds["error"]) 

465 self.assertNotIn("NumJobStarts", htc_job.cmds["log"]) 

466 self.assertTrue(htc_job.cmds["error"].endswith(".out")) 

467 self.assertTrue(htc_job.cmds["output"].endswith(".out")) 

468 self.assertTrue(htc_job.cmds["log"].endswith(".log")) 

469 

470 

471if __name__ == "__main__": 

472 unittest.main()