Coverage for tests/test_prepare

1# This file is part of ctrl_bps_htcondor.

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.

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/>.

28"""Unit tests for prepare utility functions."""

30import logging

31import os

32import unittest

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

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

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

50class TranslateJobCmdsTestCase(unittest.TestCase):

51 """Test _translate_job_cmds method."""

53 def setUp(self):

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

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

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)

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)

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})")

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))

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'")

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'")

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'")

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 def testProfileJobCommands(self):

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

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

139 gwjob.request_memory = 2048

140 gwjob.memory_multiplier = 1

141 gwjob.number_of_retries = 0

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

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

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

145

146 def testProfileCached(self):

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

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

149 gwjob.request_memory = 2048

150 gwjob.memory_multiplier = 1

151 gwjob.number_of_retries = 0

152 cached_vals = dict(self.cached_vals)

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

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

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

156

157

158class TranslateDagCmdsTestCase(unittest.TestCase):

159 """Test _translate_dag_cmds method."""

160

161 def setUp(self):

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

163

164 def testPriority(self):

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

166 gwjob.priority = 100

167 dag_commands = prepare_utils._translate_dag_cmds(gwjob)

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

169

170

171class GroupToSubdagTestCase(unittest.TestCase):

172 """Test _group_to_subdag function."""

173

174 def testBlocking(self):

175 gw = make_3_label_workflow_groups_sort("test1", True)

176 gwjob = gw.get_job("group_order1_10001")

177 config = BpsConfig(

178 {},

179 search_order=BPS_SEARCH_ORDER,

180 defaults=BPS_DEFAULTS,

181 )

182

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

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

185

186

187class GatherSiteValuesTestCase(unittest.TestCase):

188 """Test _gather_site_values function."""

189

190 def testAllThere(self):

191 config = BpsConfig(

192 {},

193 search_order=BPS_SEARCH_ORDER,

194 defaults=BPS_DEFAULTS,

195 )

196 compute_site = "notThere"

197 results = prepare_utils._gather_site_values(config, compute_site)

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

199

200 def testNotSpecified(self):

201 config = BpsConfig(

202 {},

203 search_order=BPS_SEARCH_ORDER,

204 defaults=BPS_DEFAULTS,

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

206 )

207 compute_site = "notThere"

208 results = prepare_utils._gather_site_values(config, compute_site)

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

210

211 def testGlobalNodeset(self):

212 config = BpsConfig(

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

214 search_order=BPS_SEARCH_ORDER,

215 defaults=BPS_DEFAULTS,

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

217 )

218 compute_site = "fr"

219 results = prepare_utils._gather_site_values(config, compute_site)

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

221

222 def testSiteNodeset(self):

223 config = BpsConfig(

224 {

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

226 "campaign": "DRP",

227 "site": {"fr": {"nodeset": "fr_node_set_{campaign}"}},

228 },

229 search_order=BPS_SEARCH_ORDER,

230 defaults=BPS_DEFAULTS,

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

232 )

233 compute_site = "fr"

234 results = prepare_utils._gather_site_values(config, compute_site)

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

236

237 def testAttrsProfile(self):

238 test_values = {

239 "bpsNodeset": "DEVSET",

240 "site": {

241 "mycomputer": {

242 "profile": {

243 "condor": {

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

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

246 }

247 }

248 }

249 },

250 }

251 config = BpsConfig(

252 test_values,

253 search_order=BPS_SEARCH_ORDER,

254 defaults=BPS_DEFAULTS,

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

256 )

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

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

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

260

261

262class GatherLabelValuesTestCase(unittest.TestCase):

263 """Test _gather_labels_values function."""

264

265 def testClusterLabel(self):

266 # Test cluster value overrides pipetask.

267 label = "label1"

268 config = BpsConfig(

269 {

270 "cluster": {

271 "label1": {

272 "releaseExpr": "cluster_val",

273 "overwriteJobFiles": False,

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

275 }

276 },

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

278 },

279 search_order=BPS_SEARCH_ORDER,

280 defaults=BPS_DEFAULTS,

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

282 )

283 results = prepare_utils._gather_label_values(config, label)

284 self.assertEqual(

285 results,

286 {

287 "attrs": {},

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

289 "releaseExpr": "cluster_val",

290 "overwriteJobFiles": False,

291 },

292 )

293

294 def testPipetaskLabel(self):

295 label = "label1"

296 config = BpsConfig(

297 {

298 "pipetask": {

299 "label1": {

300 "releaseExpr": "pipetask_val",

301 "overwriteJobFiles": False,

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

303 }

304 }

305 },

306 search_order=BPS_SEARCH_ORDER,

307 defaults=BPS_DEFAULTS,

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

309 )

310 results = prepare_utils._gather_label_values(config, label)

311 self.assertEqual(

312 results,

313 {

314 "attrs": {},

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

316 "releaseExpr": "pipetask_val",

317 "overwriteJobFiles": False,

318 },

319 )

320

321 def testNoSection(self):

322 label = "notThere"

323 config = BpsConfig(

324 {},

325 search_order=BPS_SEARCH_ORDER,

326 defaults=BPS_DEFAULTS,

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

328 )

329 results = prepare_utils._gather_label_values(config, label)

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

331

332 def testNoOverwriteSpecified(self):

333 label = "notthere"

334 config = BpsConfig(

335 {},

336 search_order=BPS_SEARCH_ORDER,

337 defaults={},

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

339 )

340 results = prepare_utils._gather_label_values(config, label)

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

342

343 def testFinalJob(self):

344 label = "finalJob"

345 config = BpsConfig(

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

347 search_order=BPS_SEARCH_ORDER,

348 defaults=BPS_DEFAULTS,

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

350 )

351 results = prepare_utils._gather_label_values(config, label)

352 self.assertEqual(

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

354 )

355

356

357class CreateCheckJobTestCase(unittest.TestCase):

358 """Test _create_check_job function."""

359

360 def testSuccess(self):

361 group_job_name = "group_order1_val1a"

362 job_label = "order1"

363 job = prepare_utils._create_check_job(group_job_name, job_label)

364 self.assertIn(group_job_name, job.name)

365 self.assertEqual(job.label, job_label)

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

367

368

369class CreatePeriodicReleaseExprTestCase(unittest.TestCase):

370 """Test _create_periodic_release_expr function."""

371

372 def testNoReleaseExpr(self):

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

374 self.assertEqual(results, "")

375

376 def testMultiplierNone(self):

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

378 self.assertEqual(results, "")

379

380 def testJustMemoryReleaseExpr(self):

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

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

383 truth = (

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

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

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

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

388 )

389 self.assertEqual(results, truth)

390

391 def testJustUserReleaseExpr(self):

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

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

394 self.assertEqual(results, truth)

395

396 def testJustUserReleaseExprMultiplierNone(self):

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

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

399 self.assertEqual(results, truth)

400

401 def testMemoryAndUserReleaseExpr(self):

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

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

404 truth = (

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

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

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

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

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

410 )

411 self.assertEqual(results, truth)

412

413

414class CreatePeriodicRemoveExprTestCase(unittest.TestCase):

415 """Test _create_periodic_release_expr function."""

416

417 def testBasicRemoveExpr(self):

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

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

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

421 self.assertEqual(results, truth)

422

423 def testBasicRemoveExprMultiplierNone(self):

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

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

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

427 self.assertEqual(results, truth)

428

429 def testMemoryRemoveExpr(self):

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

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

432 truth = (

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

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

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

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

437 )

438 self.assertEqual(results, truth)

439

440

441class HandleJobOutputsTestCase(unittest.TestCase):

442 """Test _handle_job_outputs function."""

443

444 def setUp(self):

445 self.job_name = "test_job"

446 self.out_prefix = "/test/prefix"

447

448 def tearDown(self):

449 pass

450

451 def testNoOutputsSharedFilesystem(self):

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

453 mock_workflow = unittest.mock.Mock()

454 mock_workflow.get_job_outputs.return_value = []

455

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

457

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

459

460 def testWithOutputsSharedFilesystem(self):

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

462 mock_workflow = unittest.mock.Mock()

463 mock_workflow.get_job_outputs.return_value = [

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

465 ]

466

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

468

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

470

471 def testNoOutputsNoSharedFilesystem(self):

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

473 mock_workflow = unittest.mock.Mock()

474 mock_workflow.get_job_outputs.return_value = []

475

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

477

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

479

480 def testWithAnOutputNoSharedFilesystem(self):

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

482 mock_workflow = unittest.mock.Mock()

483 mock_workflow.get_job_outputs.return_value = [

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

485 ]

486

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

488

489 expected = {

490 "transfer_output_files": "output.txt",

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

492 }

493 self.assertEqual(result, expected)

494

495 def testWithOutputsNoSharedFilesystem(self):

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

497 mock_workflow = unittest.mock.Mock()

498 mock_workflow.get_job_outputs.return_value = [

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

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

501 ]

502

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

504

505 expected = {

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

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

508 }

509 self.assertEqual(result, expected)

510

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

512 def testLogging(self, mock_log):

513 mock_workflow = unittest.mock.Mock()

514 mock_workflow.get_job_outputs.return_value = [

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

516 ]

517

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

519

520 self.assertTrue(mock_log.debug.called)

521 debug_calls = mock_log.debug.call_args_list

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

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

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

525

526

527class CreateJobTestCase(unittest.TestCase):

528 """Test _create_job function."""

529

530 def setUp(self):

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

532 self.template = "{label}/{tract}/{patch}/{band}/{subfilter}/{physical_filter}/{visit}/{exposure}"

533

534 def testNoOverwrite(self):

535 cached_values = {

536 "bpsUseShared": True,

537 "overwriteJobFiles": False,

538 "memoryLimit": 491520,

539 "profile": {},

540 "attrs": {},

541 }

542 gwjob = self.generic_workflow.get_final()

543 out_prefix = "submit"

544 htc_job = prepare_utils._create_job(

545 self.template, cached_values, self.generic_workflow, gwjob, out_prefix

546 )

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

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

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

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

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

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

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

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

555

556 def testNodesetWithNoRequirements(self):

557 cached_values = {

558 "bpsUseShared": True,

559 "overwriteJobFiles": False,

560 "memoryLimit": 491520,

561 "profile": {},

562 "attrs": {},

563 "nodeset": "set1",

564 }

565 gwjob = self.generic_workflow.get_job("label1_10002_11")

566 out_prefix = "temp"

567 htc_job = prepare_utils._create_job(

568 self.template, cached_values, self.generic_workflow, gwjob, out_prefix

569 )

570 self.assertEqual(htc_job.cmds["requirements"], '( Target.Nodeset == "set1" )')

571 self.assertEqual(htc_job.attrs["JobNodeset"], "set1")

572

573 def testNodesetWithRequirements(self):

574 cached_values = {

575 "bpsUseShared": True,

576 "overwriteJobFiles": False,

577 "memoryLimit": 491520,

578 "profile": {"requirements": "dummy_val == 3"},

579 "attrs": {},

580 "nodeset": "set1",

581 }

582 gwjob = self.generic_workflow.get_job("label1_10002_11")

583 out_prefix = "temp"

584 htc_job = prepare_utils._create_job(

585 self.template, cached_values, self.generic_workflow, gwjob, out_prefix

586 )

587 self.assertEqual(htc_job.cmds["requirements"], '(dummy_val == 3) && ( Target.Nodeset == "set1" )')

588 self.assertEqual(htc_job.attrs["JobNodeset"], "set1")

589

590

591if __name__ == "__main__":

592 unittest.main()

Coverage for tests / test_prepare_utils.py: 24%

283 statements