Coverage for tests/test_lssthtc.py: 19%

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

27"""Unit tests for classes and functions in lssthtc.py."""

29import io

30import logging

31import os

32import pathlib

33import stat

34import sys

35import tempfile

36import unittest

37from shutil import copy2, copytree, ignore_patterns, rmtree, which

39import htcondor

41from lsst.ctrl.bps import BpsConfig

42from lsst.ctrl.bps.htcondor import dagman_configurator, htcondor_config, lssthtc

43from lsst.daf.butler import Config

44from lsst.utils.tests import temporaryDirectory

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

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

50class TestLsstHtc(unittest.TestCase):

51 """Test basic usage."""

53 def testHtcEscapeInt(self):

54 self.assertEqual(lssthtc.htc_escape(100), 100)

56 def testHtcEscapeDouble(self):

57 self.assertEqual(lssthtc.htc_escape('"double"'), '""double""')

59 def testHtcEscapeSingle(self):

60 self.assertEqual(lssthtc.htc_escape("'single'"), "''single''")

62 def testHtcEscapeNoSideEffect(self):

63 val = "'val'"

64 self.assertEqual(lssthtc.htc_escape(val), "''val''")

65 self.assertEqual(val, "'val'")

67 def testHtcEscapeQuot(self):

68 self.assertEqual(lssthtc.htc_escape(""val""), '"val"')

70 def testHtcVersion(self):

71 ver = lssthtc.htc_version()

72 self.assertRegex(ver, r"^\d+\.\d+\.\d+$")

75class HtcTweakJobInfoTestCase(unittest.TestCase):

76 """Test the function responsible for massaging job information."""

78 def setUp(self):

79 self.log_dir = tempfile.TemporaryDirectory()

80 self.log_dirname = pathlib.Path(self.log_dir.name)

81 self.job = {

82 "Cluster": 1,

83 "Proc": 0,

84 "Iwd": str(self.log_dirname),

85 "Owner": self.log_dirname.owner(),

86 "MyType": None,

87 "TerminatedNormally": True,

88 }

90 def tearDown(self):

91 self.log_dir.cleanup()

93 def testDirectAssignments(self):

94 lssthtc.htc_tweak_log_info(self.log_dirname, self.job)

95 self.assertEqual(self.job["ClusterId"], self.job["Cluster"])

96 self.assertEqual(self.job["ProcId"], self.job["Proc"])

97 self.assertEqual(self.job["Iwd"], str(self.log_dirname))

98 self.assertEqual(self.job["Owner"], self.log_dirname.owner())

100 def testIncompatibleAdPassThru(self):

101 # Passing a job ad with insufficient information should be a no-op.

102 expected = {"foo": "bar"}

103 result = dict(expected)

104 lssthtc.htc_tweak_log_info(self.log_dirname, result)

105 self.assertEqual(result, expected)

106

107 def testJobStatusAssignmentJobAbortedEvent(self):

108 job = self.job | {"MyType": "JobAbortedEvent"}

109 lssthtc.htc_tweak_log_info(self.log_dirname, job)

110 self.assertTrue("JobStatus" in job)

111 self.assertEqual(job["JobStatus"], htcondor.JobStatus.REMOVED)

112

113 def testJobStatusAssignmentExecuteEvent(self):

114 job = self.job | {"MyType": "ExecuteEvent"}

115 lssthtc.htc_tweak_log_info(self.log_dirname, job)

116 self.assertTrue("JobStatus" in job)

117 self.assertEqual(job["JobStatus"], htcondor.JobStatus.RUNNING)

118

119 def testJobStatusAssignmentSubmitEvent(self):

120 job = self.job | {"MyType": "SubmitEvent"}

121 lssthtc.htc_tweak_log_info(self.log_dirname, job)

122 self.assertTrue("JobStatus" in job)

123 self.assertEqual(job["JobStatus"], htcondor.JobStatus.IDLE)

124

125 def testJobStatusAssignmentJobHeldEvent(self):

126 job = self.job | {"MyType": "JobHeldEvent"}

127 lssthtc.htc_tweak_log_info(self.log_dirname, job)

128 self.assertTrue("JobStatus" in job)

129 self.assertEqual(job["JobStatus"], htcondor.JobStatus.HELD)

130

131 def testJobStatusAssignmentJobTerminatedEvent(self):

132 job = self.job | {"MyType": "JobTerminatedEvent"}

133 lssthtc.htc_tweak_log_info(self.log_dirname, job)

134 self.assertTrue("JobStatus" in job)

135 self.assertEqual(job["JobStatus"], htcondor.JobStatus.COMPLETED)

136

137 def testJobStatusAssignmentPostScriptTerminatedEvent(self):

138 job = self.job | {"MyType": "PostScriptTerminatedEvent"}

139 lssthtc.htc_tweak_log_info(self.log_dirname, job)

140 self.assertTrue("JobStatus" in job)

141 self.assertEqual(job["JobStatus"], htcondor.JobStatus.COMPLETED)

142

143 def testJobStatusAssignmentReleaseEvent(self):

144 job = self.job | {"MyType": "JobReleaseEvent"}

145 lssthtc.htc_tweak_log_info(self.log_dirname, job)

146 self.assertTrue("JobStatus" in job)

147 self.assertEqual(job["JobStatus"], htcondor.JobStatus.RUNNING)

148

149 def testAddingExitStatusSuccess(self):

150 job = self.job | {

151 "MyType": "JobTerminatedEvent",

152 "ToE": {"ExitBySignal": False, "ExitCode": 1},

153 }

154 lssthtc.htc_tweak_log_info(self.log_dirname, job)

155 self.assertIn("ExitBySignal", job)

156 self.assertIs(job["ExitBySignal"], False)

157 self.assertIn("ExitCode", job)

158 self.assertEqual(job["ExitCode"], 1)

159

160 def testAddingExitStatusFailure(self):

161 job = self.job | {

162 "MyType": "JobHeldEvent",

163 }

164 with self.assertLogs(logger=logger, level="ERROR") as cm:

165 lssthtc.htc_tweak_log_info(self.log_dirname, job)

166 self.assertIn("Could not determine exit status", cm.output[0])

167

168 def testLoggingUnknownLogEvent(self):

169 job = self.job | {"MyType": "Foo"}

170 with self.assertLogs(logger=logger, level="DEBUG") as cm:

171 lssthtc.htc_tweak_log_info(self.log_dirname, job)

172 self.assertIn("Unknown log event", cm.output[1])

173

174 def testMissingKey(self):

175 job = self.job

176 del job["Cluster"]

177 with self.assertRaises(KeyError) as cm:

178 lssthtc.htc_tweak_log_info(self.log_dirname, job)

179 self.assertEqual(str(cm.exception), "'Cluster'")

180

181

182class HtcCheckDagmanOutputTestCase(unittest.TestCase):

183 """Test htc_check_dagman_output function."""

184

185 def test_missing_output_file(self):

186 with temporaryDirectory() as tmp_dir:

187 with self.assertRaises(FileNotFoundError):

188 _ = lssthtc.htc_check_dagman_output(tmp_dir)

189

190 def test_permissions_output_file(self):

191 with temporaryDirectory() as tmp_dir:

192 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir)

193 os.chmod(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out", 0o200)

194 print(os.stat(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out"))

195 results = lssthtc.htc_check_dagman_output(tmp_dir)

196 os.chmod(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out", 0o600)

197 self.assertIn("Could not read dagman output file", results)

198

199 def test_submit_failure(self):

200 with temporaryDirectory() as tmp_dir:

201 copy2(f"{TESTDIR}/data/bad_submit.dag.dagman.out", tmp_dir)

202 results = lssthtc.htc_check_dagman_output(tmp_dir)

203 self.assertIn("Warn: Job submission issues (last: ", results)

204

205 def test_tmpdir_abort(self):

206 with temporaryDirectory() as tmp_dir:

207 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir)

208 results = lssthtc.htc_check_dagman_output(tmp_dir)

209 self.assertIn("Cannot submit from /tmp", results)

210

211 def test_no_messages(self):

212 with temporaryDirectory() as tmp_dir:

213 copy2(f"{TESTDIR}/data/test_no_messages.dag.dagman.out", tmp_dir)

214 results = lssthtc.htc_check_dagman_output(tmp_dir)

215 self.assertEqual("", results)

216

217

218class SummarizeDagTestCase(unittest.TestCase):

219 """Test summarize_dag function."""

220

221 def test_no_dag_file(self):

222 with temporaryDirectory() as tmp_dir:

223 summary, job_name_to_pipetask, job_name_to_type = lssthtc.summarize_dag(tmp_dir)

224 self.assertFalse(len(job_name_to_pipetask))

225 self.assertFalse(len(job_name_to_type))

226 self.assertFalse(summary)

227

228 def test_success(self):

229 with temporaryDirectory() as tmp_dir:

230 copy2(f"{TESTDIR}/data/good.dag", tmp_dir)

231 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)

232 self.assertEqual(summary, "pipetaskInit:1;label1:1;label2:1;label3:1;finalJob:1")

233 self.assertEqual(

234 job_name_to_label,

235 {

236 "pipetaskInit": "pipetaskInit",

237 "0682f8f9-12f0-40a5-971e-8b30c7231e5c_label1_val1_val2": "label1",

238 "d0305e2d-f164-4a85-bd24-06afe6c84ed9_label2_val1_val2": "label2",

239 "2806ecc9-1bba-4362-8fff-ab4e6abb9f83_label3_val1_val2": "label3",

240 "finalJob": "finalJob",

241 },

242 )

243 self.assertEqual(

244 job_name_to_type,

245 {

246 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,

247 "0682f8f9-12f0-40a5-971e-8b30c7231e5c_label1_val1_val2": lssthtc.WmsNodeType.PAYLOAD,

248 "d0305e2d-f164-4a85-bd24-06afe6c84ed9_label2_val1_val2": lssthtc.WmsNodeType.PAYLOAD,

249 "2806ecc9-1bba-4362-8fff-ab4e6abb9f83_label3_val1_val2": lssthtc.WmsNodeType.PAYLOAD,

250 "finalJob": lssthtc.WmsNodeType.FINAL,

251 },

252 )

253

254 def test_service(self):

255 with temporaryDirectory() as tmp_dir:

256 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag", tmp_dir)

257 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)

258 self.assertEqual(summary, "pipetaskInit:1;label1:2;label2:2;finalJob:1")

259 self.assertEqual(

260 job_name_to_label,

261 {

262 "pipetaskInit": "pipetaskInit",

263 "057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b": "label1",

264 "4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a": "label1",

265 "40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b": "label2",

266 "696ee50d-e711-40d6-9caf-ee29ae4a656d_label2_val1a_val2a": "label2",

267 "finalJob": "finalJob",

268 "provisioningJob": "provisioningJob",

269 },

270 )

271 self.assertEqual(

272 job_name_to_type,

273 {

274 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,

275 "057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

276 "4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

277 "40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

278 "696ee50d-e711-40d6-9caf-ee29ae4a656d_label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

279 "finalJob": lssthtc.WmsNodeType.FINAL,

280 "provisioningJob": lssthtc.WmsNodeType.SERVICE,

281 },

282 )

283

284 def test_noop(self):

285 with temporaryDirectory() as tmp_dir:

286 copy2(f"{TESTDIR}/data/noop_running_1/noop_running_1.dag", tmp_dir)

287 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)

288 self.assertEqual(

289 set(summary.split(";")),

290 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"},

291 )

292 self.assertEqual(

293 job_name_to_label,

294 {

295 "label1_val1a_val2a": "label1",

296 "label1_val1a_val2b": "label1",

297 "label1_val1b_val2a": "label1",

298 "label1_val1b_val2b": "label1",

299 "label1_val1c_val2a": "label1",

300 "label1_val1c_val2b": "label1",

301 "label2_val1a_val2a": "label2",

302 "label2_val1a_val2b": "label2",

303 "label2_val1b_val2a": "label2",

304 "label2_val1b_val2b": "label2",

305 "label2_val1c_val2a": "label2",

306 "label2_val1c_val2b": "label2",

307 "label3_val1a_val2a": "label3",

308 "label3_val1a_val2b": "label3",

309 "label3_val1b_val2a": "label3",

310 "label3_val1b_val2b": "label3",

311 "label3_val1c_val2a": "label3",

312 "label3_val1c_val2b": "label3",

313 "label4_val1a_val2a": "label4",

314 "label4_val1a_val2b": "label4",

315 "label4_val1b_val2a": "label4",

316 "label4_val1b_val2b": "label4",

317 "label4_val1c_val2a": "label4",

318 "label4_val1c_val2b": "label4",

319 "label5_val1a_val2a": "label5",

320 "label5_val1a_val2b": "label5",

321 "label5_val1b_val2a": "label5",

322 "label5_val1b_val2b": "label5",

323 "label5_val1c_val2a": "label5",

324 "label5_val1c_val2b": "label5",

325 "finalJob": "finalJob",

326 "pipetaskInit": "pipetaskInit",

327 "wms_noop_order1_val1a": "order1",

328 "wms_noop_order1_val1b": "order1",

329 },

330 )

331 self.assertEqual(

332 job_name_to_type,

333 {

334 "label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

335 "label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

336 "label1_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

337 "label1_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

338 "label1_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

339 "label1_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

340 "label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

341 "label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

342 "label2_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

343 "label2_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

344 "label2_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

345 "label2_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

346 "label3_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

347 "label3_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

348 "label3_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

349 "label3_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

350 "label3_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

351 "label3_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

352 "label4_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

353 "label4_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

354 "label4_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

355 "label4_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

356 "label4_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

357 "label4_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

358 "label5_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

359 "label5_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

360 "label5_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

361 "label5_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

362 "label5_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

363 "label5_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

364 "finalJob": lssthtc.WmsNodeType.FINAL,

365 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,

366 "wms_noop_order1_val1a": lssthtc.WmsNodeType.NOOP,

367 "wms_noop_order1_val1b": lssthtc.WmsNodeType.NOOP,

368 },

369 )

370

371 def test_subdags(self):

372 with temporaryDirectory() as tmp_dir:

373 submit_dir = os.path.join(tmp_dir, "group_running_1")

374 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

375 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(submit_dir)

376 self.assertEqual(

377 set(summary.split(";")),

378 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"},

379 )

380

381 self.assertEqual(

382 job_name_to_label,

383 {

384 "pipetaskInit": "pipetaskInit",

385 "label1_val1b_val2a": "label1",

386 "label1_val1c_val2a": "label1",

387 "label1_val1a_val2b": "label1",

388 "label1_val1b_val2b": "label1",

389 "label1_val1c_val2b": "label1",

390 "label1_val1a_val2a": "label1",

391 "label2_val1a_val2b": "label2",

392 "label2_val1a_val2a": "label2",

393 "label2_val1b_val2a": "label2",

394 "label2_val1b_val2b": "label2",

395 "label2_val1c_val2a": "label2",

396 "label2_val1c_val2b": "label2",

397 "label3_val1b_val2a": "label3",

398 "label3_val1c_val2a": "label3",

399 "label3_val1a_val2b": "label3",

400 "label3_val1b_val2b": "label3",

401 "label3_val1c_val2b": "label3",

402 "label3_val1a_val2a": "label3",

403 "label4_val1a_val2b": "label4",

404 "label4_val1a_val2a": "label4",

405 "label4_val1b_val2a": "label4",

406 "label4_val1b_val2b": "label4",

407 "label4_val1c_val2a": "label4",

408 "label4_val1c_val2b": "label4",

409 "label5_val1a_val2b": "label5",

410 "label5_val1a_val2a": "label5",

411 "label5_val1b_val2a": "label5",

412 "label5_val1b_val2b": "label5",

413 "label5_val1c_val2a": "label5",

414 "label5_val1c_val2b": "label5",

415 "finalJob": "finalJob",

416 "provisioningJob": "provisioningJob",

417 "wms_group_order1_val1a": "order1",

418 "wms_group_order1_val1b": "order1",

419 "wms_group_order1_val1c": "order1",

420 "wms_check_status_wms_group_order1_val1a": "order1",

421 "wms_check_status_wms_group_order1_val1b": "order1",

422 "wms_check_status_wms_group_order1_val1c": "order1",

423 },

424 )

425

426 self.assertEqual(

427 job_name_to_type,

428 {

429 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,

430 "label1_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

431 "label1_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

432 "label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

433 "label1_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

434 "label1_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

435 "label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

436 "label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

437 "label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

438 "label2_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

439 "label2_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

440 "label2_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

441 "label2_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

442 "label3_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

443 "label3_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

444 "label3_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

445 "label3_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

446 "label3_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

447 "label3_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

448 "label4_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

449 "label4_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

450 "label4_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

451 "label4_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

452 "label4_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

453 "label4_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

454 "label5_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,

455 "label5_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,

456 "label5_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,

457 "label5_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,

458 "label5_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,

459 "label5_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,

460 "finalJob": lssthtc.WmsNodeType.FINAL,

461 "provisioningJob": lssthtc.WmsNodeType.SERVICE,

462 "wms_group_order1_val1a": lssthtc.WmsNodeType.SUBDAG,

463 "wms_group_order1_val1b": lssthtc.WmsNodeType.SUBDAG,

464 "wms_group_order1_val1c": lssthtc.WmsNodeType.SUBDAG,

465 "wms_check_status_wms_group_order1_val1a": lssthtc.WmsNodeType.SUBDAG_CHECK,

466 "wms_check_status_wms_group_order1_val1b": lssthtc.WmsNodeType.SUBDAG_CHECK,

467 "wms_check_status_wms_group_order1_val1c": lssthtc.WmsNodeType.SUBDAG_CHECK,

468 },

469 )

470

471

472class ReadDagNodesLogTestCase(unittest.TestCase):

473 """Test read_dag_nodes_log function."""

474

475 def setUp(self):

476 self.tmpdir = tempfile.mkdtemp()

477

478 def tearDown(self):

479 rmtree(self.tmpdir, ignore_errors=True)

480

481 def testFileMissing(self):

482 with self.assertRaisesRegex(FileNotFoundError, "DAGMan node log not found in"):

483 _ = lssthtc.read_dag_nodes_log(self.tmpdir)

484

485 def testRegular(self):

486 with temporaryDirectory() as tmp_dir:

487 submit_dir = os.path.join(tmp_dir, "tiny_problems")

488 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))

489 results = lssthtc.read_dag_nodes_log(submit_dir)

490 self.assertEqual(results["9231.0"]["Cluster"], 9231)

491 self.assertEqual(results["9231.0"]["Proc"], 0)

492 self.assertEqual(results["9231.0"]["ToE"]["ExitCode"], 1)

493 self.assertEqual(len(results), 6)

494

495 def testSubdags(self):

496 """Making sure it gets data from subdag dirs and doesn't

497 fail if some subdags haven't started running yet.

498 """

499 with temporaryDirectory() as tmp_dir:

500 submit_dir = os.path.join(tmp_dir, "group_running_1")

501 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

502 results = lssthtc.read_dag_nodes_log(submit_dir)

503 # main dag

504 self.assertEqual(results["10094.0"]["Cluster"], 10094)

505 # subdag

506 self.assertEqual(results["10112.0"]["Cluster"], 10112)

507 self.assertEqual(results["10116.0"]["Cluster"], 10116)

508

509

510class ReadNodeStatusTestCase(unittest.TestCase):

511 """Test read_node_status function."""

512

513 def setUp(self):

514 self.tmpdir = tempfile.mkdtemp()

515

516 def tearDown(self):

517 rmtree(self.tmpdir, ignore_errors=True)

518

519 def testServiceJobNotSubmitted(self):

520 # tiny_prov_no_submit files have successful workflow

521 # but provisioningJob could not submit.

522 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag.nodes.log", self.tmpdir)

523 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag.dagman.log", self.tmpdir)

524 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.node_status", self.tmpdir)

525 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag", self.tmpdir)

526

527 jobs = lssthtc.read_node_status(self.tmpdir)

528 found = [

529 id_

530 for id_ in jobs

531 if jobs[id_].get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN) == lssthtc.WmsNodeType.SERVICE

532 ]

533 self.assertEqual(len(found), 1)

534 self.assertEqual(jobs[found[0]]["DAGNodeName"], "provisioningJob")

535 self.assertEqual(jobs[found[0]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY)

536

537 def testMissingStatusFile(self):

538 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag.nodes.log", self.tmpdir)

539 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag.dagman.log", self.tmpdir)

540 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag", self.tmpdir)

541

542 jobs = lssthtc.read_node_status(self.tmpdir)

543 self.assertEqual(len(jobs), 7)

544 self.assertEqual(jobs["9230.0"]["DAGNodeName"], "pipetaskInit")

545 self.assertEqual(jobs["9230.0"]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD)

546 found = [

547 id_

548 for id_ in jobs

549 if jobs[id_].get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN) == lssthtc.WmsNodeType.SERVICE

550 ]

551 self.assertEqual(len(found), 1)

552 self.assertEqual(jobs[found[0]]["DAGNodeName"], "provisioningJob")

553

554 def testSubdagsRunning(self):

555 with temporaryDirectory() as tmp_dir:

556 test_tmp_dir = pathlib.Path(tmp_dir)

557 submit_dir = test_tmp_dir / "submit"

558 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

559 jobs = lssthtc.read_node_status(submit_dir)

560 self.assertEqual(len(jobs), 39) # includes non-payload jobs

561 # not guaranteed ids are same, so use names instead

562 job_name_to_id = {}

563 for id_, info in jobs.items():

564 job_name_to_id[info.get("DAGNodeName", id_)] = id_

565 job_type_to_names = {}

566 for id_, info in jobs.items():

567 job_type_to_names.setdefault(

568 info.get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN), set()

569 ).add(info.get("DAGNodeName", id_))

570

571 # check counts

572 self.assertNotIn(lssthtc.WmsNodeType.NOOP, job_type_to_names)

573 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.PAYLOAD]), 31)

574 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.FINAL]), 1)

575 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SERVICE]), 1)

576 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG]), 3)

577 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG_CHECK]), 3)

578

579 # spot check some statuses

580 self.assertEqual(

581 jobs[job_name_to_id["label3_val1a_val2b"]]["NodeStatus"], lssthtc.NodeStatus.DONE

582 )

583 self.assertEqual(

584 jobs[job_name_to_id["wms_group_order1_val1a"]]["NodeStatus"], lssthtc.NodeStatus.SUBMITTED

585 )

586 self.assertEqual(

587 jobs[job_name_to_id["label5_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY

588 )

589 self.assertEqual(

590 jobs[job_name_to_id["label2_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.DONE

591 )

592

593 def testSubdagsFailed(self):

594 with temporaryDirectory() as tmp_dir:

595 test_tmp_dir = pathlib.Path(tmp_dir)

596 submit_dir = test_tmp_dir / "submit"

597 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

598 jobs = lssthtc.read_node_status(submit_dir)

599 self.assertEqual(len(jobs), 39)

600 # not guaranteed ids are same, so use names instead

601 job_name_to_id = {}

602 for id_, info in jobs.items():

603 job_name_to_id[info.get("DAGNodeName", id_)] = id_

604 job_type_to_names = {}

605 for id_, info in jobs.items():

606 job_type_to_names.setdefault(

607 info.get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN), set()

608 ).add(info.get("DAGNodeName", id_))

609

610 # check counts

611 self.assertNotIn(lssthtc.WmsNodeType.NOOP, job_type_to_names)

612 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.PAYLOAD]), 31)

613 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.FINAL]), 1)

614 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SERVICE]), 1)

615 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG]), 3)

616 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG_CHECK]), 3)

617

618 # spot check some statuses

619 self.assertEqual(

620 jobs[job_name_to_id["label3_val1a_val2b"]]["NodeStatus"], lssthtc.NodeStatus.DONE

621 )

622 self.assertEqual(

623 jobs[job_name_to_id["wms_group_order1_val1a"]]["NodeStatus"], lssthtc.NodeStatus.DONE

624 )

625 self.assertEqual(

626 jobs[job_name_to_id["label5_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.DONE

627 )

628

629 self.assertEqual(

630 jobs[job_name_to_id["label5_val1b_val2a"]]["NodeStatus"], lssthtc.NodeStatus.FUTILE

631 )

632 self.assertEqual(

633 jobs[job_name_to_id["wms_group_order1_val1b"]]["NodeStatus"], lssthtc.NodeStatus.DONE

634 )

635 self.assertEqual(

636 jobs[job_name_to_id["wms_check_status_wms_group_order1_val1b"]]["NodeStatus"],

637 lssthtc.NodeStatus.ERROR,

638 )

639

640

641class HTCJobTestCase(unittest.TestCase):

642 """Test HTCJob methods."""

643

644 def testWriteDagCommandsPayload(self):

645 job = lssthtc.HTCJob(

646 "job1",

647 "label1",

648 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},

649 {"dir": "jobs/label1"},

650 )

651 job.subfile = "job1.sub"

652

653 mockfh = io.StringIO()

654 job.write_dag_commands(mockfh, "../..")

655 self.assertIn('JOB job1 "job1.sub" DIR "../../jobs/label1"', mockfh.getvalue())

656

657 def testWriteDagCommandsNotJob(self):

658 # Testing giving command_name, no dag_rel_path and no dir

659 job = lssthtc.HTCJob(

660 "finalJob",

661 "finalJob",

662 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},

663 )

664 job.subfile = "jobs/finalJob/finalJob.sub"

665 mockfh = io.StringIO()

666 job.write_dag_commands(mockfh, "", "FINAL")

667 self.assertIn('FINAL finalJob "jobs/finalJob/finalJob.sub"', mockfh.getvalue())

668

669 def testWriteDagCommandsNoop(self):

670 job = lssthtc.HTCJob("wms_noop_job1", "label1", {}, {"noop": True})

671 job.subfile = "notthere.sub"

672 mockfh = io.StringIO()

673 job.write_dag_commands(mockfh, "")

674 self.assertIn("NOOP", mockfh.getvalue())

675

676 def testWriteSubmitFile(self):

677 job = lssthtc.HTCJob(

678 "job1",

679 "label1",

680 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},

681 )

682 with temporaryDirectory() as tmp_dir:

683 filename = pathlib.Path(tmp_dir) / "label1/job1.sub"

684 job.write_submit_file(filename.parent)

685 self.assertTrue(filename.exists())

686 # Try to make Submit object from file to find any syntax issues

687 _ = lssthtc.htc_create_submit_from_file(filename)

688

689 def testWriteSubmitFileExists(self):

690 job = lssthtc.HTCJob(

691 "job1",

692 "label1",

693 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},

694 )

695 with temporaryDirectory() as tmp_dir:

696 filename = pathlib.Path(tmp_dir) / "job1.sub"

697 job.subfile = filename

698 with open(filename, "w"):

699 pass # make empty file

700 job.write_submit_file(filename.parent)

701 # make sure didn't overwrite file

702 self.assertEqual(filename.stat().st_size, 0, "Incorrectly overwrote existing file")

703

704

705class HtcWriteJobCommands(unittest.TestCase):

706 """Test _htc_write_job_commands function."""

707

708 def testAllCommands(self):

709 dag_cmds = {

710 "pre": {

711 "defer": {"status": 1, "time": 120},

712 "debug": {"filename": "debug_pre.txt", "type": "ALL"},

713 "executable": "exec1",

714 "arguments": "arg1 arg2",

715 },

716 "post": {

717 "defer": {"status": 2, "time": 180},

718 "debug": {"filename": "debug_post.txt", "type": "ALL"},

719 "executable": "exec2",

720 "arguments": "arg3 arg4",

721 },

722 "vars": {"num": 8, "spaces": "a space"},

723 "pre_skip": "1",

724 "retry": 3,

725 "retry_unless_exit": 1,

726 "abort_dag_on": {"node_exit": 100, "abort_exit": 4},

727 "priority": 123,

728 }

729

730 truth = """SCRIPT DEFER 1 120 DEBUG debug_pre.txt ALL PRE job1 exec1 arg1 arg2

731SCRIPT DEFER 2 180 DEBUG debug_post.txt ALL POST job1 exec2 arg3 arg4

732VARS job1 num="8"

733VARS job1 spaces="a space"

734PRE_SKIP job1 1

735RETRY job1 3 UNLESS-EXIT 1

736ABORT-DAG-ON job1 100 RETURN 4

737PRIORITY job1 123

738"""

739 mockfh = io.StringIO()

740 lssthtc._htc_write_job_commands(mockfh, "job1", dag_cmds)

741 self.assertEqual(mockfh.getvalue(), truth)

742

743 def testPartialCommands(self):

744 # Trigger skipping the inner if clauses.

745 dag_cmds = {

746 "pre": {

747 "executable": "exec1",

748 },

749 "post": {

750 "executable": "exec2",

751 },

752 "vars": {"num": 8, "spaces": "a space"},

753 "pre_skip": "1",

754 "retry": 3,

755 }

756

757 truth = """SCRIPT PRE job1 exec1

758SCRIPT POST job1 exec2

759VARS job1 num="8"

760VARS job1 spaces="a space"

761PRE_SKIP job1 1

762RETRY job1 3

763"""

764 mockfh = io.StringIO()

765 lssthtc._htc_write_job_commands(mockfh, "job1", dag_cmds)

766 self.assertEqual(mockfh.getvalue(), truth)

767

768 def testNoCommands(self):

769 dag_cmds = {}

770 mockfh = io.StringIO()

771 lssthtc._htc_write_job_commands(mockfh, "job2", dag_cmds)

772 self.assertEqual(mockfh.getvalue(), "")

773

774 def testFinal(self):

775 self.maxDiff = None

776 dag_cmds = {

777 "pre": {

778 "defer": {"status": 1, "time": 120},

779 "debug": {"filename": "debug_pre.txt", "type": "ALL"},

780 "executable": "exec1",

781 "arguments": "arg1 arg2",

782 },

783 "post": {

784 "defer": {"status": 2, "time": 180},

785 "debug": {"filename": "debug_post.txt", "type": "ALL"},

786 "executable": "exec2",

787 "arguments": "arg3 arg4",

788 },

789 "vars": {"num": 8, "spaces": "a space"},

790 "pre_skip": "1",

791 "retry": 3,

792 "retry_unless_exit": 1,

793 "abort_dag_on": {"node_exit": 100, "abort_exit": 4},

794 "priority": 123,

795 }

796

797 truth = """SCRIPT DEFER 1 120 DEBUG debug_pre.txt ALL PRE finalJob exec1 arg1 arg2

798SCRIPT DEFER 2 180 DEBUG debug_post.txt ALL POST finalJob exec2 arg3 arg4

799VARS finalJob num="8"

800VARS finalJob spaces="a space"

801PRE_SKIP finalJob 1

802"""

803 mockfh = io.StringIO()

804 lssthtc._htc_write_job_commands(mockfh, "finalJob", dag_cmds, "FINAL")

805 self.assertEqual(mockfh.getvalue(), truth)

806

807

808class HTCBackupFilesSinglePathTestCase(unittest.TestCase):

809 """Test htc_backup_files_single_path function."""

810

811 def testSrcDestSame(self):

812 with temporaryDirectory() as tmp_dir:

813 with self.assertRaisesRegex(

814 RuntimeError, "Destination directory is same as the source directory"

815 ):

816 lssthtc.htc_backup_files_single_path(tmp_dir, tmp_dir)

817

818 def testSuccess(self):

819 with temporaryDirectory() as tmp_dir:

820 test_tmp_dir = pathlib.Path(tmp_dir)

821 submit_dir = test_tmp_dir / "the_src_dir"

822 copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*"))

823 backup_dir = test_tmp_dir / "the_dest_dir"

824 backup_dir.mkdir()

825 lssthtc.htc_backup_files_single_path(submit_dir, backup_dir)

826 result_submit = []

827 for root, _, files in os.walk(submit_dir):

828 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])

829 self.assertEqual(

830 set(result_submit),

831 {

832 "./tiny_success.dag.dagman.log",

833 "./tiny_success.dag.dagman.out",

834 "./tiny_success.dag",

835 },

836 )

837 result_backup = []

838 for root, _, files in os.walk(backup_dir):

839 result_backup.extend([str(os.path.join(os.path.relpath(root, backup_dir), f)) for f in files])

840 self.assertEqual(

841 set(result_backup),

842 {

843 "./tiny_success.info.json",

844 "./tiny_success.dag.metrics",

845 "./tiny_success.dag.nodes.log",

846 "./tiny_success.node_status",

847 },

848 )

849

850

851class HTCBackupFilesTestCase(unittest.TestCase):

852 """Test htc_backup_files function."""

853

854 def testDirectoryNotFound(self):

855 with temporaryDirectory() as tmp_dir:

856 test_tmp_dir = pathlib.Path(tmp_dir)

857 submit_dir = test_tmp_dir / "submit"

858 with self.assertRaises(FileNotFoundError):

859 _ = lssthtc.htc_backup_files(submit_dir)

860

861 def testSuccess(self):

862 with temporaryDirectory() as tmp_dir:

863 test_tmp_dir = pathlib.Path(tmp_dir)

864 submit_dir = test_tmp_dir / "submit"

865 copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*"))

866 result_rescue = lssthtc.htc_backup_files(submit_dir)

867 self.assertIsNone(result_rescue)

868 result_submit = []

869 for root, _, files in os.walk(submit_dir):

870 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])

871 self.assertEqual(

872 set(result_submit),

873 {

874 "./tiny_success.dag.dagman.log",

875 "./tiny_success.dag.dagman.out",

876 "./tiny_success.dag",

877 "000/tiny_success.info.json",

878 "000/tiny_success.dag.metrics",

879 "000/tiny_success.dag.nodes.log",

880 "000/tiny_success.node_status",

881 },

882 )

883

884 def testDestNotInSubmitDir(self):

885 with temporaryDirectory() as tmp_dir:

886 test_tmp_dir = pathlib.Path(tmp_dir)

887 submit_dir = test_tmp_dir / "submit"

888 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))

889 with self.assertLogs("lsst.ctrl.bps.htcondor", level="WARNING") as cm:

890 result_rescue = lssthtc.htc_backup_files(submit_dir, test_tmp_dir / "backup")

891 self.assertIn("Invalid backup location:", cm.output[-1])

892 result_rescue = lssthtc.htc_backup_files(submit_dir)

893 self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue))

894 result_submit = []

895 for root, _, files in os.walk(submit_dir):

896 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])

897 self.assertEqual(

898 set(result_submit),

899 {

900 "./tiny_problems.dag.dagman.log",

901 "./tiny_problems.dag.dagman.out",

902 "./tiny_problems.dag",

903 "./tiny_problems.dag.rescue001",

904 "001/tiny_problems.info.json",

905 "001/tiny_problems.dag.metrics",

906 "001/tiny_problems.dag.nodes.log",

907 "001/tiny_problems.node_status",

908 },

909 )

910

911 def testDestInSubmitDir(self):

912 with temporaryDirectory() as tmp_dir:

913 test_tmp_dir = pathlib.Path(tmp_dir)

914 submit_dir = test_tmp_dir / "submit"

915 backup_dir = submit_dir / "subdir"

916 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))

917 result_rescue = lssthtc.htc_backup_files(submit_dir, backup_dir)

918 self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue))

919 result_submit = []

920 for root, _, files in os.walk(submit_dir):

921 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])

922 self.assertEqual(

923 set(result_submit),

924 {

925 "./tiny_problems.dag.dagman.log",

926 "./tiny_problems.dag.dagman.out",

927 "./tiny_problems.dag",

928 "./tiny_problems.dag.rescue001",

929 "subdir/001/tiny_problems.info.json",

930 "subdir/001/tiny_problems.dag.metrics",

931 "subdir/001/tiny_problems.dag.nodes.log",

932 "subdir/001/tiny_problems.node_status",

933 },

934 )

935

936 def testRelativeSubdir(self):

937 with temporaryDirectory() as tmp_dir:

938 test_tmp_dir = pathlib.Path(tmp_dir)

939 submit_dir = test_tmp_dir / "submit"

940 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))

941 result_rescue = lssthtc.htc_backup_files(submit_dir, "reldir")

942 self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue))

943 result_submit = []

944 for root, _, files in os.walk(submit_dir):

945 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])

946 self.assertEqual(

947 set(result_submit),

948 {

949 "./tiny_problems.dag.dagman.log",

950 "./tiny_problems.dag.dagman.out",

951 "./tiny_problems.dag",

952 "./tiny_problems.dag.rescue001",

953 "reldir/001/tiny_problems.info.json",

954 "reldir/001/tiny_problems.dag.metrics",

955 "reldir/001/tiny_problems.dag.nodes.log",

956 "reldir/001/tiny_problems.node_status",

957 },

958 )

959

960 def testSubdags(self):

961 with temporaryDirectory() as tmp_dir:

962 test_tmp_dir = pathlib.Path(tmp_dir)

963 submit_dir = test_tmp_dir / "submit"

964 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

965 result_rescue = lssthtc.htc_backup_files(submit_dir)

966 self.assertTrue(result_rescue.samefile(submit_dir / "group_failed_1.dag.rescue001"))

967 result_submit = []

968 for root, _, files in os.walk(submit_dir):

969 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])

970 self.assertEqual(

971 set(result_submit),

972 {

973 "./group_failed_1.dag",

974 "./group_failed_1.dag.dagman.log",

975 "./group_failed_1.dag.dagman.out",

976 "./group_failed_1.dag.rescue001",

977 "subdags/wms_group_order1_val1a/group_order1_val1a.dag",

978 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.log",

979 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.out",

980 "subdags/wms_group_order1_val1b/group_order1_val1b.dag",

981 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.log",

982 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.out",

983 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.rescue001",

984 "subdags/wms_group_order1_val1c/group_order1_val1c.dag",

985 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.log",

986 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.out",

987 "001/group_failed_1.dag.nodes.log",

988 "001/group_failed_1.info.json",

989 "001/group_failed_1.node_status",

990 "001/subdags/wms_group_order1_val1a/group_order1_val1a.dag.nodes.log",

991 "001/subdags/wms_group_order1_val1a/group_order1_val1a.node_status",

992 "001/subdags/wms_group_order1_val1a/wms_group_order1_val1a.dag.post.out",

993 "001/subdags/wms_group_order1_val1a/wms_group_order1_val1a.status.txt",

994 "001/subdags/wms_group_order1_val1b/group_order1_val1b.dag.nodes.log",

995 "001/subdags/wms_group_order1_val1b/group_order1_val1b.node_status",

996 "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.status.txt",

997 "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.dag.post.out",

998 "001/subdags/wms_group_order1_val1c/group_order1_val1c.dag.nodes.log",

999 "001/subdags/wms_group_order1_val1c/group_order1_val1c.node_status",

1000 "001/subdags/wms_group_order1_val1c/wms_group_order1_val1c.dag.post.out",

1001 "001/subdags/wms_group_order1_val1c/wms_group_order1_val1c.status.txt",

1002 },

1003 )

1004

1005

1006class UpdateRescueFileTestCase(unittest.TestCase):

1007 """Test _update_rescue_file function."""

1008

1009 def testSuccess(self):

1010 self.maxDiff = None

1011 with temporaryDirectory() as tmp_dir:

1012 test_tmp_dir = pathlib.Path(tmp_dir)

1013 submit_dir = test_tmp_dir / "submit"

1014 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

1015 rescue_file = submit_dir / "group_failed_1.dag.rescue001"

1016 lssthtc._update_rescue_file(rescue_file)

1017

1018 with open(rescue_file) as fh:

1019 lines = fh.readlines()

1020 results = "".join(lines)

1021

1022 truth = """# Rescue DAG file, created after running

1023# the u_testuser_DM-46294_group_fail_20250310T160455Z.dag DAG file

1024# Created 3/10/2025 16:08:56 UTC

1025# Rescue DAG version: 2.0.1 (partial)

1026#

1027# Total number of Nodes: 26

1028# Nodes premarked DONE: 21

1029# Nodes that failed: 2

1030# wms_group_order1_val1b,finalJob,<ENDLIST>

1031

1032DONE pipetaskInit

1033DONE label1_val1c_val2a

1034DONE label1_val1b_val2b

1035DONE label1_val1b_val2a

1036DONE label1_val1c_val2b

1037DONE label1_val1a_val2a

1038DONE label1_val1a_val2b

1039DONE label3_val1c_val2a

1040DONE label3_val1b_val2b

1041DONE label3_val1b_val2a

1042DONE label3_val1c_val2b

1043DONE label3_val1a_val2a

1044DONE label3_val1a_val2b

1045DONE wms_group_order1_val1a

1046DONE label5_val1a_val2a

1047DONE label5_val1a_val2b

1048DONE wms_group_order1_val1c

1049DONE label5_val1c_val2a

1050DONE label5_val1c_val2b

1051DONE wms_check_status_wms_group_order1_val1a

1052DONE wms_check_status_wms_group_order1_val1c

1053"""

1054

1055 print("results = ", results, file=sys.stderr)

1056 print("truth = ", truth, file=sys.stderr)

1057 self.assertEqual(results, truth)

1058

1059

1060class ReadDagStatusTestCase(unittest.TestCase):

1061 """Test read_dag_status function and read_single_dag_status."""

1062

1063 def testFileMissing(self):

1064 with temporaryDirectory() as tmp_dir:

1065 with self.assertRaisesRegex(FileNotFoundError, "DAGMan node status not found"):

1066 _ = lssthtc.read_dag_status(tmp_dir)

1067

1068 def testRegular(self):

1069 with temporaryDirectory() as tmp_dir:

1070 submit_dir = os.path.join(tmp_dir, "tiny_problems")

1071 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))

1072 results = lssthtc.read_dag_status(submit_dir)

1073 truth = {

1074 "JobProcsHeld": 0,

1075 "NodesPost": 0,

1076 "JobProcsIdle": 0,

1077 "NodesTotal": 6,

1078 "NodesFailed": 2,

1079 "NodesDone": 3,

1080 "NodesQueued": 0,

1081 "NodesPre": 0,

1082 "NodesFutile": 1,

1083 "NodesUnready": 0,

1084 }

1085 self.assertEqual(results, results | truth)

1086

1087 def testSubdags(self):

1088 """Making sure it gets data from subdag dirs and doesn't

1089 fail if some subdags haven't started running yet.

1090 """

1091 self.maxDiff = None

1092 with temporaryDirectory() as tmp_dir:

1093 submit_dir = os.path.join(tmp_dir, "submit")

1094 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))

1095 results = lssthtc.read_dag_status(submit_dir)

1096 truth = {

1097 "JobProcsHeld": 0,

1098 "NodesPost": 0,

1099 "JobProcsIdle": 0,

1100 "NodesTotal": 34,

1101 "NodesFailed": 0,

1102 "NodesDone": 17,

1103 "NodesQueued": 3,

1104 "NodesPre": 0,

1105 "NodesFutile": 0,

1106 "NodesUnready": 14,

1107 }

1108 self.assertEqual(results, results | truth)

1109

1110

1111class ReadDagInfoTestCase(unittest.TestCase):

1112 """Test read_dag_info function."""

1113

1114 def testFileMissing(self):

1115 with temporaryDirectory() as tmp_dir:

1116 with self.assertRaisesRegex(FileNotFoundError, "File with DAGMan job information not found in "):

1117 _ = lssthtc.read_dag_info(tmp_dir)

1118

1119 def testSuccess(self):

1120 with temporaryDirectory() as tmp_dir:

1121 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir)

1122 results = lssthtc.read_dag_info(tmp_dir)

1123

1124 truth = {

1125 "test02": {

1126 "9208.0": {

1127 "ClusterId": 9208,

1128 "GlobalJobId": "test02#9208.0#1739465078",

1129 "bps_wms_service": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorService",

1130 "bps_project": "dev",

1131 "bps_payload": "tiny",

1132 "bps_operator": "testuser",

1133 "bps_wms_workflow": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorWorkflow",

1134 "bps_provisioning_job": "provisioningJob",

1135 "bps_run_quanta": "label1:1;label2:1",

1136 "bps_campaign": "quick",

1137 "bps_runsite": "testpool",

1138 "bps_job_summary": "pipetaskInit:1;label1:1;label2:1;finalJob:1",

1139 "bps_run": "u_testuser_tiny_20250213T164427Z",

1140 "bps_isjob": "True",

1141 }

1142 }

1143 }

1144

1145 self.assertEqual(results, truth)

1146

1147 def testPermissionError(self):

1148 with temporaryDirectory() as tmp_dir:

1149 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir)

1150 with unittest.mock.patch("lsst.ctrl.bps.htcondor.lssthtc.open") as mocked_open:

1151 mocked_open.side_effect = PermissionError

1152 with self.assertLogs("lsst.ctrl.bps.htcondor", level="DEBUG") as cm:

1153 results = lssthtc.read_dag_info(tmp_dir)

1154 self.assertIn("Retrieving DAGMan job information failed:", cm.output[-1])

1155 self.assertEqual({}, results)

1156

1157

1158class HtcWriteCondorFileTestCase(unittest.TestCase):

1159 """Test htc_write_condor_file function."""

1160

1161 def testSuccess(self):

1162 with temporaryDirectory() as tmp_dir:

1163 job_name = "job1"

1164 filename = pathlib.Path(tmp_dir) / f"label1/{job_name}.sub"

1165 job = {

1166 "executable": "$(CTRL_MPEXEC_DIR)/bin/pipetask",

1167 "arguments": "-a -b 2 -c",

1168 "request_memory": "2000",

1169 "environment": "one=1 two=\"2\" three='spacey 'quoted' value'",

1170 "log": f"{job_name}.log",

1171 }

1172 job_attrs = {

1173 "bps_job_name": job_name,

1174 "bps_job_label": "label1",

1175 "bps_job_quanta": "task1:8;task2:8",

1176 }

1177 expected = [

1178 "executable=$(CTRL_MPEXEC_DIR)/bin/pipetask\n",

1179 "arguments=-a -b 2 -c\n",

1180 "request_memory=2000\n",

1181 "environment=\"one=1 two=\"2\" three='spacey 'quoted' value'\"\n",

1182 f"output={job_name}.$(Cluster).out\n",

1183 f"error={job_name}.$(Cluster).out\n",

1184 f"log={job_name}.log\n",

1185 f'+bps_job_name = "{job_name}"\n',

1186 '+bps_job_label = "label1"\n',

1187 '+bps_job_quanta = "task1:8;task2:8"\n',

1188 "queue\n",

1189 ]

1190

1191 lssthtc.htc_write_condor_file(filename, job_name, job, job_attrs)

1192 with open(filename, encoding="utf-8") as f:

1193 actual = f.readlines()

1194

1195 self.assertEqual(set(actual), set(expected))

1196 self.assertTrue(filename.exists())

1197 # Try to make Submit object from file to find any syntax issues

1198 _ = lssthtc.htc_create_submit_from_file(filename)

1199

1200

1201class HtcCreateSubmitFromDagTestCase(unittest.TestCase):

1202 """Test htc_create_submit_from_dag function."""

1203

1204 @classmethod

1205 def setUpClass(cls):

1206 cls.bindir = None

1207 # htcondor.Submit.from_dag requires condor_dagman executable in path.

1208 if not which("condor_dagman"): # pragma: no cover

1209 cls.bindir = tempfile.TemporaryDirectory()

1210 fake_dagman_exec = pathlib.Path(cls.bindir.name) / "condor_dagman"

1211 with open(fake_dagman_exec, "w") as fh:

1212 print("#!/bin/bash", file=fh)

1213 print("echo fake_condor_dagman $@", file=fh)

1214 print("exit 0", file=fh)

1215 fake_dagman_exec.chmod(fake_dagman_exec.stat().st_mode | stat.S_IEXEC)

1216 os.environ["PATH"] = f"{os.environ['PATH']}:{cls.bindir.name}"

1217

1218 @classmethod

1219 def tearDownClass(cls):

1220 if cls.bindir:

1221 cls.bindir.cleanup()

1222

1223 @unittest.mock.patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})

1224 def testMaxIdleEnvVar(self):

1225 with temporaryDirectory() as tmp_dir:

1226 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)

1227 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"

1228 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})

1229 self.assertIn("-MaxIdle 42", submit["arguments"])

1230

1231 @unittest.mock.patch.dict(os.environ, {})

1232 def testMaxIdleGiven(self):

1233 with temporaryDirectory() as tmp_dir:

1234 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)

1235 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"

1236 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"MaxIdle": 37})

1237 self.assertIn("-MaxIdle 37", submit["arguments"])

1238

1239 @unittest.mock.patch.dict(os.environ, {})

1240 def testNoMaxJobsIdle(self):

1241 """Note: Since the produced arguments differ depending on

1242 HTCondor version when no MaxIdle passed to from_dag, not

1243 checking arguments string here. Instead just making sure

1244 lssthtc code doesn't pass MaxIdle value to from_dag.

1245 """

1246 with temporaryDirectory() as tmp_dir:

1247 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)

1248 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"

1249 with unittest.mock.patch("htcondor.Submit.from_dag") as submit_mock:

1250 with unittest.mock.patch("htcondor.param") as mock_param:

1251 mock_param.__contains__.return_value = False

1252 _ = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})

1253 submit_mock.assert_called_once_with(str(dag_filename), {})

1254

1255

1256class HtcDagTestCase(unittest.TestCase):

1257 """Test for HTCDag class."""

1258

1259 def setUp(self):

1260 job = lssthtc.HTCJob(name="test_job")

1261 job.add_job_cmds(

1262 {

1263 "executable": "/usr/bin/echo",

1264 "arguments": "foo",

1265 "output": "test_job.$(Cluster).out",

1266 "error": "test_job.$(Cluster).out",

1267 "log": "test_job.$(Cluster).log",

1268 }

1269 )

1270 job.subfile = f"{job.name}.sub"

1271

1272 self.dag = lssthtc.HTCDag(name="test_workflow")

1273 self.dag.add_job(job)

1274

1275 self.subfile_expected = [

1276 "executable=/usr/bin/echo\n",

1277 "arguments=foo\n",

1278 "output=test_job.$(Cluster).out\n",

1279 "error=test_job.$(Cluster).out\n",

1280 "log=test_job.$(Cluster).log\n",

1281 "queue\n",

1282 ]

1283

1284 def tearDown(self):

1285 pass

1286

1287 def testWriteWithDagConfig(self):

1288 with temporaryDirectory() as tmp_dir:

1289 config = BpsConfig(Config(htcondor_config.HTC_DEFAULTS_URI))

1290 job = self.dag.nodes["test_job"]["data"]

1291 wms_config_filename = "dagman.conf"

1292 wms_configurator = dagman_configurator.DagmanConfigurator(config)

1293 wms_configurator.prepare(wms_config_filename, prefix=tmp_dir)

1294 wms_configurator.configure(self.dag)

1295 dagfile_expected = [

1296 f"CONFIG {wms_config_filename}\n",

1297 f'JOB {job.name} "{job.subfile}"\n',

1298 f"DOT {self.dag.name}.dot\n",

1299 f"NODE_STATUS_FILE {self.dag.name}.node_status\n",

1300 f'SET_JOB_ATTR bps_wms_config_path= "{wms_config_filename}"\n',

1301 ]

1302

1303 self.dag.write(tmp_dir, "", "")

1304

1305 self.assertIn("submit_path", self.dag.graph)

1306 self.assertEqual(self.dag.graph["submit_path"], tmp_dir)

1307 self.assertIn("dag_filename", self.dag.graph)

1308 self.assertEqual(self.dag.graph["dag_filename"], f"{self.dag.graph['name']}.dag")

1309 with open(os.path.join(tmp_dir, self.dag.graph["dag_filename"]), encoding="utf-8") as f:

1310 dagfile_actual = f.readlines()

1311 self.assertEqual(dagfile_actual, dagfile_expected)

1312 with open(os.path.join(tmp_dir, job.subfile), encoding="utf-8") as f:

1313 subfile_actual = f.readlines()

1314 self.assertEqual(subfile_actual, self.subfile_expected)

1315

1316 def testWriteWithoutDagConfig(self):

1317 with temporaryDirectory() as tmp_dir:

1318 job = self.dag.nodes["test_job"]["data"]

1319 dagfile_expected = [

1320 f'JOB {job.name} "{job.subfile}"\n',

1321 f"DOT {self.dag.name}.dot\n",

1322 f"NODE_STATUS_FILE {self.dag.name}.node_status\n",

1323 ]

1324

1325 self.dag.write(tmp_dir, "", "")

1326

1327 self.assertIn("submit_path", self.dag.graph)

1328 self.assertEqual(self.dag.graph["submit_path"], tmp_dir)

1329 self.assertIn("dag_filename", self.dag.graph)

1330 self.assertEqual(self.dag.graph["dag_filename"], f"{self.dag.graph['name']}.dag")

1331 with open(os.path.join(tmp_dir, self.dag.graph["dag_filename"]), encoding="utf-8") as f:

1332 dagfile_actual = f.readlines()

1333 self.assertEqual(dagfile_actual, dagfile_expected)

1334 with open(os.path.join(tmp_dir, job.subfile), encoding="utf-8") as f:

1335 subfile_actual = f.readlines()

1336 self.assertEqual(subfile_actual, self.subfile_expected)

1337

1338

1339if __name__ == "__main__":

1340 unittest.main()

Coverage for tests / test_lssthtc.py: 19%

565 statements