Coverage for tests/test_prepare_utils.py: 100%

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 shutil 

33import unittest 

34from copy import deepcopy 

35 

36from networkx import is_isomorphic 

37 

38from lsst.ctrl.bps import ( 

39 BPS_DEFAULTS, 

40 BPS_SEARCH_ORDER, 

41 BpsConfig, 

42 GenericWorkflow, 

43 GenericWorkflowExec, 

44 GenericWorkflowFile, 

45 GenericWorkflowJob, 

46) 

47from lsst.ctrl.bps.htcondor import lssthtc, prepare_utils 

48from lsst.ctrl.bps.htcondor.htcondor_config import HTC_DEFAULTS_URI 

49from lsst.ctrl.bps.tests.gw_test_utils import ( 

50 make_3_label_workflow, 

51 make_3_label_workflow_groups_sort, 

52 make_lazy_workflow, 

53) 

54from lsst.daf.butler import Config 

55from lsst.utils.tests import temporaryDirectory 

56 

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

58 

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

60 

61 

62class TranslateJobCmdsTestCase(unittest.TestCase): 

63 """Test _translate_job_cmds method.""" 

64 

65 def setUp(self): 

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

67 self.cached_vals = { 

68 "profile": {}, 

69 "bpsUseShared": True, 

70 "memoryLimit": 32768, 

71 "bpsMakeCommand": True, 

72 "bpsUseHTCEnvironment": True, 

73 } 

74 

75 def testRetryUnlessNone(self): 

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

77 gwjob.retry_unless_exit = None 

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

79 self.assertNotIn("retry_until", htc_commands) 

80 

81 def testRetryUnlessInt(self): 

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

83 gwjob.retry_unless_exit = 3 

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

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

86 

87 def testRetryUnlessList(self): 

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

89 gwjob.retry_unless_exit = [1, 2] 

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

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

92 

93 def testRetryUnlessBad(self): 

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

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

96 with self.assertRaises(ValueError) as cm: 

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

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

99 

100 def testEnvironmentBasic(self): 

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

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

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

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

105 

106 def testEnvironmentSpaces(self): 

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

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

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

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

111 

112 def testEnvironmentSingleQuotes(self): 

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

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

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

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

117 

118 def testEnvironmentDoubleQuotes(self): 

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

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

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

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

123 

124 def testEnvironmentWithEnvVars(self): 

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

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

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

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

129 

130 def testPeriodicRelease(self): 

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

132 gwjob.request_memory = 2048 

133 gwjob.memory_multiplier = 2 

134 gwjob.number_of_retries = 3 

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

136 release = ( 

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

138 "(HoldReasonCode =?= 12 || (HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " 

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

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

141 ) 

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

143 

144 def testPeriodicRemoveNoRetries(self): 

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

146 gwjob.request_memory = 2048 

147 gwjob.memory_multiplier = 1 

148 gwjob.number_of_retries = 0 

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

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

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

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

153 

154 def testProfileJobCommands(self): 

155 requirement_str = 'Machine == "node01.cluster.local"' 

156 gwjob = GenericWorkflowJob("requirements", "label1", executable=self.gw_exec) 

157 gwjob.request_memory = 2048 

158 gwjob.memory_multiplier = 1 

159 gwjob.number_of_retries = 0 

160 gwjob.profile = {"requirements": requirement_str} 

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

162 self.assertEqual(htc_commands["requirements"], requirement_str) 

163 

164 def testProfileCached(self): 

165 requirement_str = 'Machine == "node01.cluster.local"' 

166 gwjob = GenericWorkflowJob("requirements", "label1", executable=self.gw_exec) 

167 gwjob.request_memory = 2048 

168 gwjob.memory_multiplier = 1 

169 gwjob.number_of_retries = 0 

170 cached_vals = dict(self.cached_vals) 

171 cached_vals["profile"] = {"requirements": requirement_str} 

172 htc_commands = prepare_utils._translate_job_cmds(cached_vals, None, gwjob) 

173 self.assertEqual(htc_commands["requirements"], requirement_str) 

174 

175 def testArgumentsReplaceWmsVars(self): 

176 gwjob = GenericWorkflowJob("job1", "label1", executable=self.gw_exec) 

177 gw = GenericWorkflow("test1") 

178 gw.add_job(gwjob) 

179 gwjob.request_cpus = 1 

180 gwjob.request_memory = 2048 

181 gwjob.arguments = "run-qbb repo test.qg --summary /a/b/t/jobs/c/d/job-<WMS:attemptNum>-summary.json" 

182 new_arguments = "run-qbb repo test.qg --summary /a/b/t/jobs/c/d/job-$$([NumJobStarts])-summary.json" 

183 htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, gw, gwjob) 

184 self.assertEqual(htc_commands["arguments"], new_arguments) 

185 

186 

187class TranslateCommandLineTestCase(unittest.TestCase): 

188 """Test _translate_command_line method.""" 

189 

190 def setUp(self): 

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

192 self.cached_vals = {"bpsUseShared": True, "bpsMakeCommand": True, "bpsUseHTCEnvironment": True} 

193 

194 def _make_job(self, name="job1", executable=None, arguments=None): 

195 gwjob = GenericWorkflowJob(name, "label1", executable=executable or self.gw_exec) 

196 gw = GenericWorkflow("test1") 

197 gw.add_job(gwjob) 

198 if arguments is not None: 

199 gwjob.arguments = arguments 

200 return gw, gwjob 

201 

202 def testMakeCommandBasic(self): 

203 # Default bpsMakeCommand (True), no executable transfer, no arguments. 

204 gw, gwjob = self._make_job() 

205 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

206 self.assertEqual(jobcmds["getenv"], "True") 

207 self.assertEqual(jobcmds["executable"], "/dummy/dir/pipetask") 

208 self.assertNotIn("transfer_executable", jobcmds) 

209 self.assertNotIn("arguments", jobcmds) 

210 

211 def testMakeCommandTransferExecutable(self): 

212 gw_exec = GenericWorkflowExec("test_exec", "/dummy/dir/pipetask", transfer_executable=True) 

213 gw, gwjob = self._make_job(executable=gw_exec) 

214 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

215 self.assertEqual(jobcmds["transfer_executable"], "True") 

216 self.assertEqual(jobcmds["executable"], "/dummy/dir/pipetask") 

217 

218 def testMakeCommandExecutableEnvVar(self): 

219 # Environment placeholders in the executable are converted to HTCondor 

220 # env syntax when the executable is not transferred. 

221 gw_exec = GenericWorkflowExec("test_exec", "<ENV:CTRL_BPS_DIR>/bin/pipetask") 

222 gw, gwjob = self._make_job(executable=gw_exec) 

223 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

224 self.assertEqual(jobcmds["executable"], "$ENV(CTRL_BPS_DIR)/bin/pipetask") 

225 

226 def testMakeCommandBareExecutableResolvedViaPath(self): 

227 # A bare executable name (no path, no env var) is resolved to its 

228 # full path via PATH because HTCondor resolves the executable 

229 # relative to the job sandbox, not via PATH. 

230 resolved = shutil.which("sh") 

231 self.assertIsNotNone(resolved, "test requires 'sh' on PATH") 

232 gw_exec = GenericWorkflowExec("sh", "sh") 

233 gw, gwjob = self._make_job(executable=gw_exec) 

234 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

235 self.assertEqual(jobcmds["executable"], resolved) 

236 

237 def testMakeCommandBareExecutableNotOnPath(self): 

238 # A bare executable name that cannot be found on PATH is left as is. 

239 gw_exec = GenericWorkflowExec("nosuchcmd", "nosuchcmd_xyzzy") 

240 gw, gwjob = self._make_job(executable=gw_exec) 

241 with self.assertLogs(level="WARNING") as cm_log: 

242 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

243 self.assertRegex(cm_log.output[0], "Could not resolve executable 'nosuchcmd_xyzzy' on PATH.*") 

244 self.assertEqual(jobcmds["executable"], "nosuchcmd_xyzzy") 

245 

246 def testMakeCommandArguments(self): 

247 # Arguments should have cmd, wms, file, and env placeholders replaced. 

248 gw, gwjob = self._make_job(arguments="run <FILE:qg> --attempt <WMS:attemptNum> {opt}") 

249 gwjob.cmdvals = {"opt": "X"} 

250 gwfile = GenericWorkflowFile("qg", src_uri="/path/to/test.qg", wms_transfer=True, job_shared=True) 

251 gw.add_job_inputs(gwjob.name, [gwfile]) 

252 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

253 self.assertEqual(jobcmds["arguments"], "run /path/to/test.qg --attempt $$([NumJobStarts]) X") 

254 

255 def testPayloadCommandBasic(self): 

256 # bpsMakeCommand False wraps the payloadCommand in /bin/bash -c. 

257 gw, gwjob = self._make_job(arguments="run -b repo") 

258 cached_vals = { 

259 "bpsUseShared": True, 

260 "bpsMakeCommand": False, 

261 "payloadCommand": "setup; {gwjobCommand}", 

262 } 

263 jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) 

264 self.assertEqual(jobcmds["executable"], "/bin/bash") 

265 self.assertEqual(jobcmds["transfer_executable"], "False") 

266 self.assertNotIn("getenv", jobcmds) 

267 self.assertEqual(jobcmds["arguments"], "-c 'setup; /dummy/dir/pipetask run -b repo'") 

268 

269 def testPayloadCommandNoArguments(self): 

270 # No argument to the gwjob command 

271 gw, gwjob = self._make_job(arguments="") 

272 cached_vals = { 

273 "bpsUseShared": True, 

274 "bpsMakeCommand": False, 

275 "payloadCommand": "setup; {gwjobCommand}", 

276 } 

277 jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) 

278 self.assertEqual(jobcmds["executable"], "/bin/bash") 

279 self.assertEqual(jobcmds["transfer_executable"], "False") 

280 self.assertNotIn("getenv", jobcmds) 

281 self.assertEqual(jobcmds["arguments"], "-c 'setup; /dummy/dir/pipetask '") 

282 

283 def testPayloadCommandStripsNewlines(self): 

284 gw, gwjob = self._make_job(arguments="run") 

285 cached_vals = { 

286 "bpsUseShared": True, 

287 "bpsMakeCommand": False, 

288 "payloadCommand": "setup;\n{gwjobCommand}", 

289 } 

290 jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) 

291 self.assertEqual(jobcmds["arguments"], "-c 'setup;/dummy/dir/pipetask run'") 

292 

293 def testPayloadCommandExecutableEnvVar(self): 

294 # Environment placeholders in the executable use shell syntax here. 

295 gw_exec = GenericWorkflowExec("test_exec", "<ENV:CTRL_BPS_DIR>/bin/pipetask") 

296 gw, gwjob = self._make_job(executable=gw_exec, arguments="go") 

297 cached_vals = {"bpsUseShared": True, "bpsMakeCommand": False, "payloadCommand": "{gwjobCommand}"} 

298 jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) 

299 self.assertEqual(jobcmds["arguments"], "-c '${CTRL_BPS_DIR}/bin/pipetask go'") 

300 

301 def testPayloadCommandTransferExecutable(self): 

302 # Transferred executable is added to the job inputs and chmod'd. 

303 gw_exec = GenericWorkflowExec("test_exec", "/dummy/dir/pipetask", transfer_executable=True) 

304 gw, gwjob = self._make_job(executable=gw_exec, arguments="sub") 

305 cached_vals = {"bpsUseShared": True, "bpsMakeCommand": False, "payloadCommand": "{gwjobCommand}"} 

306 jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) 

307 self.assertEqual(jobcmds["arguments"], "-c 'chmod u+x pipetask; ./pipetask sub'") 

308 input_names = [f.name for f in gw.get_job_inputs(gwjob.name, data=True)] 

309 self.assertIn("test_exec", input_names) 

310 

311 def testEnvironment(self): 

312 gw, gwjob = self._make_job() 

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

314 jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) 

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

316 

317 def testPayloadCommandEnvironmentShell(self): 

318 # Exports in commands, no environment in jobcmds 

319 gw_exec = GenericWorkflowExec("test_exec", "<ENV:CTRL_BPS_DIR>/bin/pipetask") 

320 gw, gwjob = self._make_job(executable=gw_exec, arguments="go") 

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

322 cached_vals = { 

323 "bpsUseShared": True, 

324 "bpsMakeCommand": False, 

325 "bpsUseHTCEnvironment": False, 

326 "payloadCommand": "{gwjobExports} {gwjobCommand}", 

327 } 

328 jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) 

329 self.assertIn("export TEST_ENV_VAR='${CTRL_BPS_DIR}/tests';", jobcmds["arguments"]) 

330 self.assertNotIn("environment", jobcmds) 

331 

332 

333class TranslateDagCmdsTestCase(unittest.TestCase): 

334 """Test _translate_dag_cmds method.""" 

335 

336 def setUp(self): 

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

338 

339 def testPriority(self): 

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

341 gwjob.priority = 100 

342 dag_commands = prepare_utils._translate_dag_cmds(gwjob) 

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

344 

345 

346class GroupToSubdagTestCase(unittest.TestCase): 

347 """Test _group_to_subdag function.""" 

348 

349 def testBlocking(self): 

350 gw = make_3_label_workflow_groups_sort("test1", True) 

351 gwjob = gw.get_job("group_order1_10001") 

352 config = BpsConfig( 

353 {}, 

354 search_order=BPS_SEARCH_ORDER, 

355 defaults=BPS_DEFAULTS, 

356 ) 

357 

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

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

360 

361 

362class GatherSiteValuesTestCase(unittest.TestCase): 

363 """Test _gather_site_values function.""" 

364 

365 def testAllThere(self): 

366 config = BpsConfig( 

367 {}, 

368 search_order=BPS_SEARCH_ORDER, 

369 defaults=BPS_DEFAULTS, 

370 ) 

371 compute_site = "notThere" 

372 results = prepare_utils._gather_site_values(config, compute_site) 

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

374 

375 def testNotSpecified(self): 

376 config = BpsConfig( 

377 {}, 

378 search_order=BPS_SEARCH_ORDER, 

379 defaults=BPS_DEFAULTS, 

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

381 ) 

382 compute_site = "notThere" 

383 results = prepare_utils._gather_site_values(config, compute_site) 

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

385 

386 def testAttrsProfile(self): 

387 test_values = { 

388 "bpsNodeset": "DEVSET", 

389 "site": { 

390 "mycomputer": { 

391 "profile": { 

392 "condor": { 

393 "requirements": '( TARGET.Nodeset == "{bpsNodeset}" )', 

394 "+JobNodeset": "{bpsNodeset}", 

395 } 

396 } 

397 } 

398 }, 

399 } 

400 config = BpsConfig( 

401 test_values, 

402 search_order=BPS_SEARCH_ORDER, 

403 defaults=BPS_DEFAULTS, 

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

405 ) 

406 results = prepare_utils._gather_site_values(config, "mycomputer") 

407 self.assertEqual(results["profile"], {"requirements": '( TARGET.Nodeset == "DEVSET" )'}) 

408 self.assertEqual(results["attrs"], {"JobNodeset": "DEVSET"}) 

409 

410 

411class GatherLabelValuesTestCase(unittest.TestCase): 

412 """Test _gather_labels_values function.""" 

413 

414 def testClusterLabel(self): 

415 # Test cluster value overrides pipetask. 

416 config = BpsConfig( 

417 { 

418 "cluster": { 

419 "label1": { 

420 "releaseExpr": "cluster_val", 

421 "overwriteJobFiles": False, 

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

423 } 

424 }, 

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

426 "site": {"site1": {}}, 

427 }, 

428 search_order=BPS_SEARCH_ORDER, 

429 defaults=BPS_DEFAULTS, 

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

431 ) 

432 results = prepare_utils._gather_label_values(config, "label1") 

433 self.assertEqual( 

434 results, 

435 { 

436 "attrs": {}, 

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

438 "releaseExpr": "cluster_val", 

439 "overwriteJobFiles": False, 

440 "bpsMakeCommand": True, 

441 "bpsUseHTCEnvironment": True, 

442 "bpsUseShared": True, 

443 "memoryLimit": 491520, 

444 }, 

445 ) 

446 

447 def testPipetaskLabel(self): 

448 label = "label1" 

449 config = BpsConfig( 

450 { 

451 "pipetask": { 

452 "label1": { 

453 "releaseExpr": "pipetask_val", 

454 "overwriteJobFiles": False, 

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

456 } 

457 }, 

458 "site": {"site1": {}}, 

459 }, 

460 search_order=BPS_SEARCH_ORDER, 

461 defaults=BPS_DEFAULTS, 

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

463 ) 

464 results = prepare_utils._gather_label_values(config, label) 

465 self.assertEqual( 

466 results, 

467 { 

468 "attrs": {}, 

469 "bpsMakeCommand": True, 

470 "bpsUseHTCEnvironment": True, 

471 "bpsUseShared": True, 

472 "memoryLimit": 491520, 

473 "overwriteJobFiles": False, 

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

475 "releaseExpr": "pipetask_val", 

476 }, 

477 ) 

478 

479 def testNoSection(self): 

480 label = "notThere" 

481 config = BpsConfig( 

482 {"site": {"site1": {}}}, 

483 search_order=BPS_SEARCH_ORDER, 

484 defaults=BPS_DEFAULTS, 

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

486 ) 

487 results = prepare_utils._gather_label_values(config, label) 

488 self.assertEqual( 

489 results, 

490 { 

491 "attrs": {}, 

492 "profile": {}, 

493 "overwriteJobFiles": True, 

494 "bpsMakeCommand": True, 

495 "bpsUseHTCEnvironment": True, 

496 "bpsUseShared": True, 

497 "memoryLimit": 491520, 

498 }, 

499 ) 

500 

501 def testNoOverwriteSpecified(self): 

502 label = "notthere" 

503 config = BpsConfig( 

504 {"site": {"site1": {}}, "memoryLimit": 491520}, 

505 search_order=BPS_SEARCH_ORDER, 

506 defaults={}, 

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

508 ) 

509 results = prepare_utils._gather_label_values(config, label) 

510 self.assertEqual( 

511 results, 

512 { 

513 "attrs": {}, 

514 "profile": {}, 

515 "overwriteJobFiles": True, 

516 "bpsMakeCommand": True, 

517 "bpsUseHTCEnvironment": True, 

518 "bpsUseShared": False, 

519 "memoryLimit": 491520, 

520 }, 

521 ) 

522 

523 def testFinalJob(self): 

524 label = "finalJob" 

525 config = BpsConfig( 

526 {"site": {"site1": {}}, "finalJob": {"profile": {"condor": {"prof_val2": 6, "+attr_val1": 5}}}}, 

527 search_order=BPS_SEARCH_ORDER, 

528 defaults=BPS_DEFAULTS, 

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

530 ) 

531 results = prepare_utils._gather_label_values(config, label) 

532 self.assertEqual( 

533 results, 

534 { 

535 "attrs": {"attr_val1": 5}, 

536 "profile": {"prof_val2": 6}, 

537 "overwriteJobFiles": False, 

538 "bpsMakeCommand": True, 

539 "bpsUseHTCEnvironment": True, 

540 "bpsUseShared": True, 

541 "memoryLimit": 491520, 

542 }, 

543 ) 

544 

545 def testGlobalNodeset(self): 

546 config = BpsConfig( 

547 {"nodeset": "global_node_set_{campaign}", "campaign": "DRP"}, 

548 search_order=BPS_SEARCH_ORDER, 

549 defaults=BPS_DEFAULTS, 

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

551 ) 

552 results = prepare_utils._gather_label_values(config, "label1") 

553 self.assertEqual(results["nodeset"], "global_node_set_DRP") 

554 

555 def testSiteNodeset(self): 

556 config = BpsConfig( 

557 { 

558 "nodeset": "global_node_set_{campaign}", 

559 "campaign": "DRP", 

560 "site": {"fr": {"nodeset": "fr_node_set_{campaign}", "siteVar": "frSiteVal"}}, 

561 "computeSite": "fr", 

562 }, 

563 search_order=BPS_SEARCH_ORDER, 

564 defaults=BPS_DEFAULTS, 

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

566 ) 

567 results = prepare_utils._gather_label_values(config, "label1") 

568 self.assertEqual(results["nodeset"], "fr_node_set_DRP") 

569 self.assertEqual(results["siteVar"], "frSiteVal") 

570 

571 def testBpsMakeCommandFalse(self): 

572 config = BpsConfig( 

573 { 

574 "bpsMakeCommand": False, 

575 }, 

576 search_order=BPS_SEARCH_ORDER, 

577 defaults=BPS_DEFAULTS, 

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

579 ) 

580 results = prepare_utils._gather_label_values(config, "label1") 

581 self.assertIn("payloadCommand", results) 

582 self.assertIn("gwjobCommand", results["payloadCommand"]) 

583 

584 

585class CreateCheckJobTestCase(unittest.TestCase): 

586 """Test _create_check_job function.""" 

587 

588 def testSuccess(self): 

589 group_job_name = "group_order1_val1a" 

590 job_label = "order1" 

591 job = prepare_utils._create_check_job(group_job_name, job_label, {}) 

592 self.assertIn(group_job_name, job.name) 

593 self.assertEqual(job.label, job_label) 

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

595 self.assertNotIn("job_nodeset", job.dagcmds["vars"]) 

596 

597 def testNodeSetSuccess(self): 

598 group_job_name = "group_order1_val1a" 

599 job_label = "order1" 

600 job = prepare_utils._create_check_job(group_job_name, job_label, {"nodeset": "custom_nodeset"}) 

601 self.assertIn(group_job_name, job.name) 

602 self.assertEqual(job.label, job_label) 

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

604 self.assertIn("job_nodeset", job.dagcmds["vars"]) 

605 

606 

607class CreatePeriodicReleaseExprTestCase(unittest.TestCase): 

608 """Test _create_periodic_release_expr function.""" 

609 

610 def setUp(self): 

611 self.maxDiff = None 

612 

613 def testNoReleaseExpr(self): 

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

615 self.assertEqual(results, "") 

616 

617 def testMultiplierNone(self): 

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

619 self.assertEqual(results, "") 

620 

621 def testJustMemoryReleaseExpr(self): 

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

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

624 truth = ( 

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

626 "(HoldReasonCode =?= 12 || " 

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

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

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

630 ) 

631 self.assertEqual(results, truth) 

632 

633 def testJustUserReleaseExpr(self): 

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

635 truth = ( 

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

637 "(HoldReasonCode =?= 12 || HoldReasonCode =!= 1 && True)" 

638 ) 

639 self.assertEqual(results, truth) 

640 

641 def testJustUserReleaseExprMultiplierNone(self): 

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

643 truth = ( 

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

645 "(HoldReasonCode =?= 12 || HoldReasonCode =!= 1 && True)" 

646 ) 

647 self.assertEqual(results, truth) 

648 

649 def testMemoryAndUserReleaseExpr(self): 

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

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

652 truth = ( 

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

654 "(HoldReasonCode =?= 12 || (HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " 

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

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

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

658 ) 

659 self.assertEqual(results, truth) 

660 

661 

662class CreatePeriodicRemoveExprTestCase(unittest.TestCase): 

663 """Test _create_periodic_release_expr function.""" 

664 

665 def testBasicRemoveExpr(self): 

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

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

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

669 self.assertEqual(results, truth) 

670 

671 def testBasicRemoveExprMultiplierNone(self): 

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

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

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

675 self.assertEqual(results, truth) 

676 

677 def testMemoryRemoveExpr(self): 

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

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

680 truth = ( 

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

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

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

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

685 ) 

686 self.assertEqual(results, truth) 

687 

688 

689class HandleJobOutputsTestCase(unittest.TestCase): 

690 """Test _handle_job_outputs function.""" 

691 

692 def setUp(self): 

693 self.job_name = "test_job" 

694 self.out_prefix = "/test/prefix" 

695 

696 def tearDown(self): 

697 pass 

698 

699 def testNoOutputsSharedFilesystem(self): 

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

701 mock_workflow = unittest.mock.Mock() 

702 mock_workflow.get_job_outputs.return_value = [] 

703 

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

705 

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

707 

708 def testWithOutputsSharedFilesystem(self): 

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

710 mock_workflow = unittest.mock.Mock() 

711 mock_workflow.get_job_outputs.return_value = [ 

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

713 ] 

714 

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

716 

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

718 

719 def testNoOutputsNoSharedFilesystem(self): 

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

721 mock_workflow = unittest.mock.Mock() 

722 mock_workflow.get_job_outputs.return_value = [] 

723 

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

725 

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

727 

728 def testWithAnOutputNoSharedFilesystem(self): 

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

730 mock_workflow = unittest.mock.Mock() 

731 mock_workflow.get_job_outputs.return_value = [ 

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

733 ] 

734 

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

736 

737 expected = { 

738 "transfer_output_files": "output.txt", 

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

740 } 

741 self.assertEqual(result, expected) 

742 

743 def testWithOutputsNoSharedFilesystem(self): 

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

745 mock_workflow = unittest.mock.Mock() 

746 mock_workflow.get_job_outputs.return_value = [ 

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

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

749 ] 

750 

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

752 

753 expected = { 

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

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

756 } 

757 self.assertEqual(result, expected) 

758 

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

760 def testLogging(self, mock_log): 

761 mock_workflow = unittest.mock.Mock() 

762 mock_workflow.get_job_outputs.return_value = [ 

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

764 ] 

765 

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

767 

768 self.assertTrue(mock_log.debug.called)