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 

35from unittest.mock import patch 

36 

37from networkx import is_isomorphic 

38 

39from lsst.ctrl.bps import ( 

40 BPS_DEFAULTS, 

41 BPS_SEARCH_ORDER, 

42 BpsConfig, 

43 GenericWorkflow, 

44 GenericWorkflowExec, 

45 GenericWorkflowFile, 

46 GenericWorkflowJob, 

47) 

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

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

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

51 make_3_label_workflow, 

52 make_3_label_workflow_groups_sort, 

53 make_lazy_workflow, 

54) 

55from lsst.daf.butler import Config 

56from lsst.utils.tests import temporaryDirectory 

57 

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

59 

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

61 

62 

63class TranslateJobCmdsTestCase(unittest.TestCase): 

64 """Test _translate_job_cmds method.""" 

65 

66 def setUp(self): 

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

68 self.cached_vals = { 

69 "profile": {}, 

70 "bpsUseShared": True, 

71 "memoryLimit": 32768, 

72 "bpsMakeCommand": True, 

73 "bpsUseHTCEnvironment": True, 

74 } 

75 

76 def testRetryUnlessNone(self): 

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

78 gwjob.retry_unless_exit = None 

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

80 self.assertNotIn("retry_until", htc_commands) 

81 

82 def testRetryUnlessInt(self): 

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

84 gwjob.retry_unless_exit = 3 

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

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

87 

88 def testRetryUnlessList(self): 

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

90 gwjob.retry_unless_exit = [1, 2] 

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

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

93 

94 def testRetryUnlessBad(self): 

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

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

97 with self.assertRaises(ValueError) as cm: 

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

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

100 

101 def testEnvironmentBasic(self): 

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

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

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

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

106 

107 def testEnvironmentSpaces(self): 

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

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

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

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

112 

113 def testEnvironmentSingleQuotes(self): 

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

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

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

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

118 

119 def testEnvironmentDoubleQuotes(self): 

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

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

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

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

124 

125 def testEnvironmentWithEnvVars(self): 

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

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

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

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

130 

131 def testPeriodicRelease(self): 

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

133 gwjob.request_memory = 2048 

134 gwjob.memory_multiplier = 2 

135 gwjob.number_of_retries = 3 

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

137 release = ( 

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

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

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

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

142 ) 

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

144 

145 def testPeriodicRemoveNoRetries(self): 

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

147 gwjob.request_memory = 2048 

148 gwjob.memory_multiplier = 1 

149 gwjob.number_of_retries = 0 

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

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

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

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

154 

155 def testProfileJobCommands(self): 

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

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

158 gwjob.request_memory = 2048 

159 gwjob.memory_multiplier = 1 

160 gwjob.number_of_retries = 0 

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

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

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

164 

165 def testProfileCached(self): 

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

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

168 gwjob.request_memory = 2048 

169 gwjob.memory_multiplier = 1 

170 gwjob.number_of_retries = 0 

171 cached_vals = dict(self.cached_vals) 

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

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

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

175 

176 def testArgumentsReplaceWmsVars(self): 

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

178 gw = GenericWorkflow("test1") 

179 gw.add_job(gwjob) 

180 gwjob.request_cpus = 1 

181 gwjob.request_memory = 2048 

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

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

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

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

186 

187 

188class TranslateCommandLineTestCase(unittest.TestCase): 

189 """Test _translate_command_line method.""" 

190 

191 def setUp(self): 

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

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

194 

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

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

197 gw = GenericWorkflow("test1") 

198 gw.add_job(gwjob) 

199 if arguments is not None: 

200 gwjob.arguments = arguments 

201 return gw, gwjob 

202 

203 def testMakeCommandBasic(self): 

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

205 gw, gwjob = self._make_job() 

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

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

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

209 self.assertNotIn("transfer_executable", jobcmds) 

210 self.assertNotIn("arguments", jobcmds) 

211 

212 def testMakeCommandTransferExecutable(self): 

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

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

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

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

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

218 

219 def testMakeCommandExecutableEnvVar(self): 

220 # Environment placeholders in the executable are converted to HTCondor 

221 # env syntax when the executable is not transferred. 

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

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

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

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

226 

227 def testMakeCommandBareExecutableResolvedViaPath(self): 

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

229 # full path via PATH because HTCondor resolves the executable 

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

231 resolved = shutil.which("sh") 

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

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

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

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

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

237 

238 def testMakeCommandBareExecutableNotOnPath(self): 

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

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

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

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

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

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

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

246 

247 def testMakeCommandArguments(self): 

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

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

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

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

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

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

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

255 

256 def testPayloadCommandBasic(self): 

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

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

259 cached_vals = { 

260 "bpsUseShared": True, 

261 "bpsMakeCommand": False, 

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

263 } 

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

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

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

267 self.assertNotIn("getenv", jobcmds) 

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

269 

270 def testPayloadCommandNoArguments(self): 

271 # No argument to the gwjob command 

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

273 cached_vals = { 

274 "bpsUseShared": True, 

275 "bpsMakeCommand": False, 

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

277 } 

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

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

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

281 self.assertNotIn("getenv", jobcmds) 

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

283 

284 def testPayloadCommandStripsNewlines(self): 

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

286 cached_vals = { 

287 "bpsUseShared": True, 

288 "bpsMakeCommand": False, 

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

290 } 

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

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

293 

294 def testPayloadCommandExecutableEnvVar(self): 

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

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

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

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

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

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

301 

302 def testPayloadCommandTransferExecutable(self): 

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

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

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

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

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

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

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

310 self.assertIn("test_exec", input_names) 

311 

312 def testEnvironment(self): 

313 gw, gwjob = self._make_job() 

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

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

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

317 

318 def testPayloadCommandEnvironmentShell(self): 

319 # Exports in commands, no environment in jobcmds 

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

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

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

323 cached_vals = { 

324 "bpsUseShared": True, 

325 "bpsMakeCommand": False, 

326 "bpsUseHTCEnvironment": False, 

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

328 } 

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

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

331 self.assertNotIn("environment", jobcmds) 

332 

333 

334class TranslateDagCmdsTestCase(unittest.TestCase): 

335 """Test _translate_dag_cmds method.""" 

336 

337 def setUp(self): 

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

339 

340 def testPriority(self): 

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

342 gwjob.priority = 100 

343 dag_commands = prepare_utils._translate_dag_cmds(gwjob) 

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

345 

346 

347class GroupToSubdagTestCase(unittest.TestCase): 

348 """Test _group_to_subdag function.""" 

349 

350 def testBlocking(self): 

351 gw = make_3_label_workflow_groups_sort("test1", True) 

352 gwjob = gw.get_job("group_order1_10001") 

353 config = BpsConfig( 

354 {}, 

355 search_order=BPS_SEARCH_ORDER, 

356 defaults=BPS_DEFAULTS, 

357 ) 

358 

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

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

361 

362 

363class GatherSiteValuesTestCase(unittest.TestCase): 

364 """Test _gather_site_values function.""" 

365 

366 def testAllThere(self): 

367 config = BpsConfig( 

368 {}, 

369 search_order=BPS_SEARCH_ORDER, 

370 defaults=BPS_DEFAULTS, 

371 ) 

372 compute_site = "notThere" 

373 results = prepare_utils._gather_site_values(config, compute_site) 

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

375 

376 def testNotSpecified(self): 

377 config = BpsConfig( 

378 {}, 

379 search_order=BPS_SEARCH_ORDER, 

380 defaults=BPS_DEFAULTS, 

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

382 ) 

383 compute_site = "notThere" 

384 results = prepare_utils._gather_site_values(config, compute_site) 

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

386 

387 def testAttrsProfile(self): 

388 test_values = { 

389 "bpsNodeset": "DEVSET", 

390 "site": { 

391 "mycomputer": { 

392 "profile": { 

393 "condor": { 

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

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

396 } 

397 } 

398 } 

399 }, 

400 } 

401 config = BpsConfig( 

402 test_values, 

403 search_order=BPS_SEARCH_ORDER, 

404 defaults=BPS_DEFAULTS, 

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

406 ) 

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

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

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

410 

411 

412class GatherLabelValuesTestCase(unittest.TestCase): 

413 """Test _gather_labels_values function.""" 

414 

415 def testClusterLabel(self): 

416 # Test cluster value overrides pipetask. 

417 config = BpsConfig( 

418 { 

419 "cluster": { 

420 "label1": { 

421 "releaseExpr": "cluster_val", 

422 "overwriteJobFiles": False, 

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

424 } 

425 }, 

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

427 "site": {"site1": {}}, 

428 }, 

429 search_order=BPS_SEARCH_ORDER, 

430 defaults=BPS_DEFAULTS, 

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

432 ) 

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

434 self.assertEqual( 

435 results, 

436 { 

437 "attrs": {}, 

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

439 "releaseExpr": "cluster_val", 

440 "overwriteJobFiles": False, 

441 "bpsMakeCommand": True, 

442 "bpsUseHTCEnvironment": True, 

443 "bpsUseShared": True, 

444 "memoryLimit": 491520, 

445 }, 

446 ) 

447 

448 def testPipetaskLabel(self): 

449 label = "label1" 

450 config = BpsConfig( 

451 { 

452 "pipetask": { 

453 "label1": { 

454 "releaseExpr": "pipetask_val", 

455 "overwriteJobFiles": False, 

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

457 } 

458 }, 

459 "site": {"site1": {}}, 

460 }, 

461 search_order=BPS_SEARCH_ORDER, 

462 defaults=BPS_DEFAULTS, 

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

464 ) 

465 results = prepare_utils._gather_label_values(config, label) 

466 self.assertEqual( 

467 results, 

468 { 

469 "attrs": {}, 

470 "bpsMakeCommand": True, 

471 "bpsUseHTCEnvironment": True, 

472 "bpsUseShared": True, 

473 "memoryLimit": 491520, 

474 "overwriteJobFiles": False, 

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

476 "releaseExpr": "pipetask_val", 

477 }, 

478 ) 

479 

480 def testNoSection(self): 

481 label = "notThere" 

482 config = BpsConfig( 

483 {"site": {"site1": {}}}, 

484 search_order=BPS_SEARCH_ORDER, 

485 defaults=BPS_DEFAULTS, 

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

487 ) 

488 results = prepare_utils._gather_label_values(config, label) 

489 self.assertEqual( 

490 results, 

491 { 

492 "attrs": {}, 

493 "profile": {}, 

494 "overwriteJobFiles": True, 

495 "bpsMakeCommand": True, 

496 "bpsUseHTCEnvironment": True, 

497 "bpsUseShared": True, 

498 "memoryLimit": 491520, 

499 }, 

500 ) 

501 

502 def testNoOverwriteSpecified(self): 

503 label = "notthere" 

504 config = BpsConfig( 

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

506 search_order=BPS_SEARCH_ORDER, 

507 defaults={}, 

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

509 ) 

510 results = prepare_utils._gather_label_values(config, label) 

511 self.assertEqual( 

512 results, 

513 { 

514 "attrs": {}, 

515 "profile": {}, 

516 "overwriteJobFiles": True, 

517 "bpsMakeCommand": True, 

518 "bpsUseHTCEnvironment": True, 

519 "bpsUseShared": False, 

520 "memoryLimit": 491520, 

521 }, 

522 ) 

523 

524 def testFinalJob(self): 

525 label = "finalJob" 

526 config = BpsConfig( 

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

528 search_order=BPS_SEARCH_ORDER, 

529 defaults=BPS_DEFAULTS, 

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

531 ) 

532 results = prepare_utils._gather_label_values(config, label) 

533 self.assertEqual( 

534 results, 

535 { 

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

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

538 "overwriteJobFiles": False, 

539 "bpsMakeCommand": True, 

540 "bpsUseHTCEnvironment": True, 

541 "bpsUseShared": True, 

542 "memoryLimit": 491520, 

543 }, 

544 ) 

545 

546 def testGlobalNodeset(self): 

547 config = BpsConfig( 

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

549 search_order=BPS_SEARCH_ORDER, 

550 defaults=BPS_DEFAULTS, 

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

552 ) 

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

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

555 

556 def testSiteNodeset(self): 

557 config = BpsConfig( 

558 { 

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

560 "campaign": "DRP", 

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

562 "computeSite": "fr", 

563 }, 

564 search_order=BPS_SEARCH_ORDER, 

565 defaults=BPS_DEFAULTS, 

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

567 ) 

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

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

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

571 

572 def testBpsMakeCommandFalse(self): 

573 config = BpsConfig( 

574 { 

575 "bpsMakeCommand": False, 

576 }, 

577 search_order=BPS_SEARCH_ORDER, 

578 defaults=BPS_DEFAULTS, 

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

580 ) 

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

582 self.assertIn("payloadCommand", results) 

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

584 

585 

586class CreateCheckJobTestCase(unittest.TestCase): 

587 """Test _create_check_job function.""" 

588 

589 def testSuccess(self): 

590 group_job_name = "group_order1_val1a" 

591 job_label = "order1" 

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

593 self.assertIn(group_job_name, job.name) 

594 self.assertEqual(job.label, job_label) 

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

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

597 

598 def testNodeSetSuccess(self): 

599 group_job_name = "group_order1_val1a" 

600 job_label = "order1" 

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

602 self.assertIn(group_job_name, job.name) 

603 self.assertEqual(job.label, job_label) 

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

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

606 

607 

608class CreatePeriodicReleaseExprTestCase(unittest.TestCase): 

609 """Test _create_periodic_release_expr function.""" 

610 

611 def setUp(self): 

612 self.maxDiff = None 

613 

614 def testNoReleaseExpr(self): 

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

616 self.assertEqual(results, "") 

617 

618 def testMultiplierNone(self): 

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

620 self.assertEqual(results, "") 

621 

622 def testJustMemoryReleaseExpr(self): 

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

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

625 truth = ( 

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

627 "(HoldReasonCode =?= 12 || " 

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

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

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

631 ) 

632 self.assertEqual(results, truth) 

633 

634 def testJustUserReleaseExpr(self): 

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

636 truth = ( 

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

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

639 ) 

640 self.assertEqual(results, truth) 

641 

642 def testJustUserReleaseExprMultiplierNone(self): 

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

644 truth = ( 

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

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

647 ) 

648 self.assertEqual(results, truth) 

649 

650 def testMemoryAndUserReleaseExpr(self): 

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

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

653 truth = ( 

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

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

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

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

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

659 ) 

660 self.assertEqual(results, truth) 

661 

662 

663class CreatePeriodicRemoveExprTestCase(unittest.TestCase): 

664 """Test _create_periodic_release_expr function.""" 

665 

666 def testBasicRemoveExpr(self): 

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

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

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

670 self.assertEqual(results, truth) 

671 

672 def testBasicRemoveExprMultiplierNone(self): 

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

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

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

676 self.assertEqual(results, truth) 

677 

678 def testMemoryRemoveExpr(self): 

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

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

681 truth = ( 

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

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

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

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

686 ) 

687 self.assertEqual(results, truth) 

688 

689 

690class HandleJobOutputsTestCase(unittest.TestCase): 

691 """Test _handle_job_outputs function.""" 

692 

693 def setUp(self): 

694 self.job_name = "test_job" 

695 self.out_prefix = "/test/prefix" 

696 

697 def tearDown(self): 

698 pass 

699 

700 def testNoOutputsSharedFilesystem(self): 

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

702 mock_workflow = unittest.mock.Mock() 

703 mock_workflow.get_job_outputs.return_value = [] 

704 

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

706 

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

708 

709 def testWithOutputsSharedFilesystem(self): 

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

711 mock_workflow = unittest.mock.Mock() 

712 mock_workflow.get_job_outputs.return_value = [ 

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

714 ] 

715 

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

717 

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

719 

720 def testNoOutputsNoSharedFilesystem(self): 

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

722 mock_workflow = unittest.mock.Mock() 

723 mock_workflow.get_job_outputs.return_value = [] 

724 

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

726 

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

728 

729 def testWithAnOutputNoSharedFilesystem(self): 

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

731 mock_workflow = unittest.mock.Mock() 

732 mock_workflow.get_job_outputs.return_value = [ 

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

734 ] 

735 

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

737 

738 expected = { 

739 "transfer_output_files": "output.txt", 

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

741 } 

742 self.assertEqual(result, expected) 

743 

744 def testWithOutputsNoSharedFilesystem(self): 

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

746 mock_workflow = unittest.mock.Mock() 

747 mock_workflow.get_job_outputs.return_value = [ 

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

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

750 ] 

751 

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

753 

754 expected = { 

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

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

757 } 

758 self.assertEqual(result, expected) 

759 

760 @patch("lsst.ctrl.bps.htcondor.prepare_utils._LOG") 

761 def testLogging(self, mock_log): 

762 mock_workflow = unittest.mock.Mock() 

763 mock_workflow.get_job_outputs.return_value = [ 

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

765 ] 

766 

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

768 

769 self.assertTrue(mock_log.debug.called)