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 TranslateDagCmdsTestCase(unittest.TestCase): 

138 """Test _translate_dag_cmds method.""" 

139 

140 def setUp(self): 

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

142 

143 def testPriority(self): 

144 gwjob = GenericWorkflowJob("priority", "label1", executable=self.gw_exec) 

145 gwjob.priority = 100 

146 dag_commands = prepare_utils._translate_dag_cmds(gwjob) 

147 self.assertEqual(dag_commands["priority"], 100) 

148 

149 

150class GroupToSubdagTestCase(unittest.TestCase): 

151 """Test _group_to_subdag function.""" 

152 

153 def testBlocking(self): 

154 gw = make_3_label_workflow_groups_sort("test1", True) 

155 gwjob = gw.get_job("group_order1_10001") 

156 config = BpsConfig( 

157 {}, 

158 search_order=BPS_SEARCH_ORDER, 

159 defaults=BPS_DEFAULTS, 

160 ) 

161 

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

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

164 

165 

166class GatherSiteValuesTestCase(unittest.TestCase): 

167 """Test _gather_site_values function.""" 

168 

169 def testAllThere(self): 

170 config = BpsConfig( 

171 {}, 

172 search_order=BPS_SEARCH_ORDER, 

173 defaults=BPS_DEFAULTS, 

174 ) 

175 compute_site = "notThere" 

176 results = prepare_utils._gather_site_values(config, compute_site) 

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

178 

179 def testNotSpecified(self): 

180 config = BpsConfig( 

181 {}, 

182 search_order=BPS_SEARCH_ORDER, 

183 defaults=BPS_DEFAULTS, 

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

185 ) 

186 compute_site = "notThere" 

187 results = prepare_utils._gather_site_values(config, compute_site) 

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

189 

190 

191class GatherLabelValuesTestCase(unittest.TestCase): 

192 """Test _gather_labels_values function.""" 

193 

194 def testClusterLabel(self): 

195 # Test cluster value overrides pipetask. 

196 label = "label1" 

197 config = BpsConfig( 

198 { 

199 "cluster": { 

200 "label1": { 

201 "releaseExpr": "cluster_val", 

202 "overwriteJobFiles": False, 

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

204 } 

205 }, 

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

207 }, 

208 search_order=BPS_SEARCH_ORDER, 

209 defaults=BPS_DEFAULTS, 

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

211 ) 

212 results = prepare_utils._gather_label_values(config, label) 

213 self.assertEqual( 

214 results, 

215 { 

216 "attrs": {}, 

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

218 "releaseExpr": "cluster_val", 

219 "overwriteJobFiles": False, 

220 }, 

221 ) 

222 

223 def testPipetaskLabel(self): 

224 label = "label1" 

225 config = BpsConfig( 

226 { 

227 "pipetask": { 

228 "label1": { 

229 "releaseExpr": "pipetask_val", 

230 "overwriteJobFiles": False, 

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

232 } 

233 } 

234 }, 

235 search_order=BPS_SEARCH_ORDER, 

236 defaults=BPS_DEFAULTS, 

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

238 ) 

239 results = prepare_utils._gather_label_values(config, label) 

240 self.assertEqual( 

241 results, 

242 { 

243 "attrs": {}, 

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

245 "releaseExpr": "pipetask_val", 

246 "overwriteJobFiles": False, 

247 }, 

248 ) 

249 

250 def testNoSection(self): 

251 label = "notThere" 

252 config = BpsConfig( 

253 {}, 

254 search_order=BPS_SEARCH_ORDER, 

255 defaults=BPS_DEFAULTS, 

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

257 ) 

258 results = prepare_utils._gather_label_values(config, label) 

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

260 

261 def testNoOverwriteSpecified(self): 

262 label = "notthere" 

263 config = BpsConfig( 

264 {}, 

265 search_order=BPS_SEARCH_ORDER, 

266 defaults={}, 

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

268 ) 

269 results = prepare_utils._gather_label_values(config, label) 

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

271 

272 def testFinalJob(self): 

273 label = "finalJob" 

274 config = BpsConfig( 

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

276 search_order=BPS_SEARCH_ORDER, 

277 defaults=BPS_DEFAULTS, 

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

279 ) 

280 results = prepare_utils._gather_label_values(config, label) 

281 self.assertEqual( 

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

283 ) 

284 

285 

286class CreateCheckJobTestCase(unittest.TestCase): 

287 """Test _create_check_job function.""" 

288 

289 def testSuccess(self): 

290 group_job_name = "group_order1_val1a" 

291 job_label = "order1" 

292 job = prepare_utils._create_check_job(group_job_name, job_label) 

293 self.assertIn(group_job_name, job.name) 

294 self.assertEqual(job.label, job_label) 

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

296 

297 

298class CreatePeriodicReleaseExprTestCase(unittest.TestCase): 

299 """Test _create_periodic_release_expr function.""" 

300 

301 def testNoReleaseExpr(self): 

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

303 self.assertEqual(results, "") 

304 

305 def testMultiplierNone(self): 

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

307 self.assertEqual(results, "") 

308 

309 def testJustMemoryReleaseExpr(self): 

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

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

312 truth = ( 

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

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

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

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

317 ) 

318 self.assertEqual(results, truth) 

319 

320 def testJustUserReleaseExpr(self): 

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

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

323 self.assertEqual(results, truth) 

324 

325 def testJustUserReleaseExprMultiplierNone(self): 

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

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

328 self.assertEqual(results, truth) 

329 

330 def testMemoryAndUserReleaseExpr(self): 

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

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

333 truth = ( 

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

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

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

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

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

339 ) 

340 self.assertEqual(results, truth) 

341 

342 

343class CreatePeriodicRemoveExprTestCase(unittest.TestCase): 

344 """Test _create_periodic_release_expr function.""" 

345 

346 def testBasicRemoveExpr(self): 

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

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

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

350 self.assertEqual(results, truth) 

351 

352 def testBasicRemoveExprMultiplierNone(self): 

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

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

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

356 self.assertEqual(results, truth) 

357 

358 def testMemoryRemoveExpr(self): 

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

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

361 truth = ( 

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

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

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

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

366 ) 

367 self.assertEqual(results, truth) 

368 

369 

370class HandleJobOutputsTestCase(unittest.TestCase): 

371 """Test _handle_job_outputs function.""" 

372 

373 def setUp(self): 

374 self.job_name = "test_job" 

375 self.out_prefix = "/test/prefix" 

376 

377 def tearDown(self): 

378 pass 

379 

380 def testNoOutputsSharedFilesystem(self): 

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

382 mock_workflow = unittest.mock.Mock() 

383 mock_workflow.get_job_outputs.return_value = [] 

384 

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

386 

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

388 

389 def testWithOutputsSharedFilesystem(self): 

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

391 mock_workflow = unittest.mock.Mock() 

392 mock_workflow.get_job_outputs.return_value = [ 

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

394 ] 

395 

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

397 

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

399 

400 def testNoOutputsNoSharedFilesystem(self): 

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

402 mock_workflow = unittest.mock.Mock() 

403 mock_workflow.get_job_outputs.return_value = [] 

404 

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

406 

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

408 

409 def testWithAnOutputNoSharedFilesystem(self): 

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

411 mock_workflow = unittest.mock.Mock() 

412 mock_workflow.get_job_outputs.return_value = [ 

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

414 ] 

415 

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

417 

418 expected = { 

419 "transfer_output_files": "output.txt", 

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

421 } 

422 self.assertEqual(result, expected) 

423 

424 def testWithOutputsNoSharedFilesystem(self): 

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

426 mock_workflow = unittest.mock.Mock() 

427 mock_workflow.get_job_outputs.return_value = [ 

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

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

430 ] 

431 

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

433 

434 expected = { 

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

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

437 } 

438 self.assertEqual(result, expected) 

439 

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

441 def testLogging(self, mock_log): 

442 mock_workflow = unittest.mock.Mock() 

443 mock_workflow.get_job_outputs.return_value = [ 

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

445 ] 

446 

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

448 

449 self.assertTrue(mock_log.debug.called) 

450 debug_calls = mock_log.debug.call_args_list 

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

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

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

454 

455 

456class CreateJobTestCase(unittest.TestCase): 

457 """Test _create_job function.""" 

458 

459 def setUp(self): 

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

461 

462 def testNoOverwrite(self): 

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

464 cached_values = { 

465 "bpsUseShared": True, 

466 "overwriteJobFiles": False, 

467 "memoryLimit": 491520, 

468 "profile": {}, 

469 "attrs": {}, 

470 } 

471 gwjob = self.generic_workflow.get_final() 

472 out_prefix = "submit" 

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

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

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

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

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

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

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

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

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

482 

483 

484if __name__ == "__main__": 

485 unittest.main()