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

775class HTCBackupFilesSinglePathTestCase(unittest.TestCase):

776 """Test htc_backup_files_single_path function."""

777

778 def testSrcDestSame(self):

779 with temporaryDirectory() as tmp_dir:

780 with self.assertRaisesRegex(

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

782 ):

783 lssthtc.htc_backup_files_single_path(tmp_dir, tmp_dir)

784

785 def testSuccess(self):

786 with temporaryDirectory() as tmp_dir:

787 test_tmp_dir = pathlib.Path(tmp_dir)

788 submit_dir = test_tmp_dir / "the_src_dir"

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

790 backup_dir = test_tmp_dir / "the_dest_dir"

791 backup_dir.mkdir()

792 lssthtc.htc_backup_files_single_path(submit_dir, backup_dir)

793 result_submit = []

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

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

796 self.assertEqual(

797 set(result_submit),

798 {

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

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

801 "./tiny_success.dag",

802 },

803 )

804 result_backup = []

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

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

807 self.assertEqual(

808 set(result_backup),

809 {

810 "./tiny_success.info.json",

811 "./tiny_success.dag.metrics",

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

813 "./tiny_success.node_status",

814 },

815 )

816

817

818class HTCBackupFilesTestCase(unittest.TestCase):

819 """Test htc_backup_files function."""

820

821 def testDirectoryNotFound(self):

822 with temporaryDirectory() as tmp_dir:

823 test_tmp_dir = pathlib.Path(tmp_dir)

824 submit_dir = test_tmp_dir / "submit"

825 with self.assertRaises(FileNotFoundError):

826 _ = lssthtc.htc_backup_files(submit_dir)

827

828 def testSuccess(self):

829 with temporaryDirectory() as tmp_dir:

830 test_tmp_dir = pathlib.Path(tmp_dir)

831 submit_dir = test_tmp_dir / "submit"

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

833 result_rescue = lssthtc.htc_backup_files(submit_dir)

834 self.assertIsNone(result_rescue)

835 result_submit = []

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

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

838 self.assertEqual(

839 set(result_submit),

840 {

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

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

843 "./tiny_success.dag",

844 "000/tiny_success.info.json",

845 "000/tiny_success.dag.metrics",

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

847 "000/tiny_success.node_status",

848 },

849 )

850

851 def testDestNotInSubmitDir(self):

852 with temporaryDirectory() as tmp_dir:

853 test_tmp_dir = pathlib.Path(tmp_dir)

854 submit_dir = test_tmp_dir / "submit"

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

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

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

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

859 result_rescue = lssthtc.htc_backup_files(submit_dir)

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

861 result_submit = []

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

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

864 self.assertEqual(

865 set(result_submit),

866 {

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

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

869 "./tiny_problems.dag",

870 "./tiny_problems.dag.rescue001",

871 "001/tiny_problems.info.json",

872 "001/tiny_problems.dag.metrics",

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

874 "001/tiny_problems.node_status",

875 },

876 )

877

878 def testDestInSubmitDir(self):

879 with temporaryDirectory() as tmp_dir:

880 test_tmp_dir = pathlib.Path(tmp_dir)

881 submit_dir = test_tmp_dir / "submit"

882 backup_dir = submit_dir / "subdir"

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

884 result_rescue = lssthtc.htc_backup_files(submit_dir, backup_dir)

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

886 result_submit = []

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

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

889 self.assertEqual(

890 set(result_submit),

891 {

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

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

894 "./tiny_problems.dag",

895 "./tiny_problems.dag.rescue001",

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

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

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

899 "subdir/001/tiny_problems.node_status",

900 },

901 )

902

903 def testRelativeSubdir(self):

904 with temporaryDirectory() as tmp_dir:

905 test_tmp_dir = pathlib.Path(tmp_dir)

906 submit_dir = test_tmp_dir / "submit"

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

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

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

910 result_submit = []

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

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

913 self.assertEqual(

914 set(result_submit),

915 {

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

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

918 "./tiny_problems.dag",

919 "./tiny_problems.dag.rescue001",

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

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

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

923 "reldir/001/tiny_problems.node_status",

924 },

925 )

926

927 def testSubdags(self):

928 with temporaryDirectory() as tmp_dir:

929 test_tmp_dir = pathlib.Path(tmp_dir)

930 submit_dir = test_tmp_dir / "submit"

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

932 result_rescue = lssthtc.htc_backup_files(submit_dir)

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

934 result_submit = []

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

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

937 self.assertEqual(

938 set(result_submit),

939 {

940 "./group_failed_1.dag",

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

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

943 "./group_failed_1.dag.rescue001",

944 "subdags/wms_group_order1_val1a/group_order1_val1a.dag",

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

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

947 "subdags/wms_group_order1_val1b/group_order1_val1b.dag",

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

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

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

951 "subdags/wms_group_order1_val1c/group_order1_val1c.dag",

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

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

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

955 "001/group_failed_1.info.json",

956 "001/group_failed_1.node_status",

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

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

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

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

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

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

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

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

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

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

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

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

969 },

970 )

971

972

973class UpdateRescueFileTestCase(unittest.TestCase):

974 """Test _update_rescue_file function."""

975

976 def testSuccess(self):

977 self.maxDiff = None

978 with temporaryDirectory() as tmp_dir:

979 test_tmp_dir = pathlib.Path(tmp_dir)

980 submit_dir = test_tmp_dir / "submit"

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

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

983 lssthtc._update_rescue_file(rescue_file)

984

985 with open(rescue_file) as fh:

986 lines = fh.readlines()

987 results = "".join(lines)

988

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

990# the u_testuser_DM-46294_group_fail_20250310T160455Z.dag DAG file

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

992# Rescue DAG version: 2.0.1 (partial)

993#

994# Total number of Nodes: 26

995# Nodes premarked DONE: 21

996# Nodes that failed: 2

997# wms_group_order1_val1b,finalJob,<ENDLIST>

998

999DONE pipetaskInit

1000DONE label1_val1c_val2a

1001DONE label1_val1b_val2b

1002DONE label1_val1b_val2a

1003DONE label1_val1c_val2b

1004DONE label1_val1a_val2a

1005DONE label1_val1a_val2b

1006DONE label3_val1c_val2a

1007DONE label3_val1b_val2b

1008DONE label3_val1b_val2a

1009DONE label3_val1c_val2b

1010DONE label3_val1a_val2a

1011DONE label3_val1a_val2b

1012DONE wms_group_order1_val1a

1013DONE label5_val1a_val2a

1014DONE label5_val1a_val2b

1015DONE wms_group_order1_val1c

1016DONE label5_val1c_val2a

1017DONE label5_val1c_val2b

1018DONE wms_check_status_wms_group_order1_val1a

1019DONE wms_check_status_wms_group_order1_val1c

1020"""

1021

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

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

1024 self.assertEqual(results, truth)

1025

1026

1027class ReadDagStatusTestCase(unittest.TestCase):

1028 """Test read_dag_status function and read_single_dag_status."""

1029

1030 def testFileMissing(self):

1031 with temporaryDirectory() as tmp_dir:

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

1033 _ = lssthtc.read_dag_status(tmp_dir)

1034

1035 def testRegular(self):

1036 with temporaryDirectory() as tmp_dir:

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

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

1039 results = lssthtc.read_dag_status(submit_dir)

1040 truth = {

1041 "JobProcsHeld": 0,

1042 "NodesPost": 0,

1043 "JobProcsIdle": 0,

1044 "NodesTotal": 6,

1045 "NodesFailed": 2,

1046 "NodesDone": 3,

1047 "NodesQueued": 0,

1048 "NodesPre": 0,

1049 "NodesFutile": 1,

1050 "NodesUnready": 0,

1051 }

1052 self.assertEqual(results, results | truth)

1053

1054 def testSubdags(self):

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

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

1057 """

1058 self.maxDiff = None

1059 with temporaryDirectory() as tmp_dir:

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

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

1062 results = lssthtc.read_dag_status(submit_dir)

1063 truth = {

1064 "JobProcsHeld": 0,

1065 "NodesPost": 0,

1066 "JobProcsIdle": 0,

1067 "NodesTotal": 34,

1068 "NodesFailed": 0,

1069 "NodesDone": 17,

1070 "NodesQueued": 3,

1071 "NodesPre": 0,

1072 "NodesFutile": 0,

1073 "NodesUnready": 14,

1074 }

1075 self.assertEqual(results, results | truth)

1076

1077

1078class ReadDagInfoTestCase(unittest.TestCase):

1079 """Test read_dag_info function."""

1080

1081 def testFileMissing(self):

1082 with temporaryDirectory() as tmp_dir:

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

1084 _ = lssthtc.read_dag_info(tmp_dir)

1085

1086 def testSuccess(self):

1087 with temporaryDirectory() as tmp_dir:

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

1089 results = lssthtc.read_dag_info(tmp_dir)

1090

1091 truth = {

1092 "test02": {

1093 "9208.0": {

1094 "ClusterId": 9208,

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

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

1097 "bps_project": "dev",

1098 "bps_payload": "tiny",

1099 "bps_operator": "testuser",

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

1101 "bps_provisioning_job": "provisioningJob",

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

1103 "bps_campaign": "quick",

1104 "bps_runsite": "testpool",

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

1106 "bps_run": "u_testuser_tiny_20250213T164427Z",

1107 "bps_isjob": "True",

1108 }

1109 }

1110 }

1111

1112 self.assertEqual(results, truth)

1113

1114 def testPermissionError(self):

1115 with temporaryDirectory() as tmp_dir:

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

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

1118 mocked_open.side_effect = PermissionError

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

1120 results = lssthtc.read_dag_info(tmp_dir)

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

1122 self.assertEqual({}, results)

1123

1124

1125class HtcWriteCondorFileTestCase(unittest.TestCase):

1126 """Test htc_write_condor_file function."""

1127

1128 def testSuccess(self):

1129 with temporaryDirectory() as tmp_dir:

1130 job_name = "job1"

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

1132 job = {

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

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

1135 "request_memory": "2000",

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

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

1138 }

1139 job_attrs = {

1140 "bps_job_name": job_name,

1141 "bps_job_label": "label1",

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

1143 }

1144 expected = [

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

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

1147 "request_memory=2000\n",

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

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

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

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

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

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

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

1155 "queue\n",

1156 ]

1157

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

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

1160 actual = f.readlines()

1161

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

1163 self.assertTrue(filename.exists())

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

1165 _ = lssthtc.htc_create_submit_from_file(filename)

1166

1167

1168class HtcCreateSubmitFromDagTestCase(unittest.TestCase):

1169 """Test htc_create_submit_from_dag function."""

1170

1171 @classmethod

1172 def setUpClass(cls):

1173 cls.bindir = None

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

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

1176 cls.bindir = tempfile.TemporaryDirectory()

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

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

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

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

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

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

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

1184

1185 @classmethod

1186 def tearDownClass(cls):

1187 if cls.bindir:

1188 cls.bindir.cleanup()

1189

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

1191 def testMaxIdleEnvVar(self):

1192 with temporaryDirectory() as tmp_dir:

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

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

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

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

1197

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

1199 def testMaxIdleGiven(self):

1200 with temporaryDirectory() as tmp_dir:

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

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

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

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

1205

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

1207 def testNoMaxJobsIdle(self):

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

1209 HTCondor version when no MaxIdle passed to from_dag, not

1210 checking arguments string here. Instead just making sure

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

1212 """

1213 with temporaryDirectory() as tmp_dir:

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

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

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

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

1218 mock_param.__contains__.return_value = False

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

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

1221

1222

1223class HtcDagTestCase(unittest.TestCase):

1224 """Test for HTCDag class."""

1225

1226 def setUp(self):

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

1228 job.add_job_cmds(

1229 {

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

1231 "arguments": "foo",

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

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

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

1235 }

1236 )

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

1238

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

1240 self.dag.add_job(job)

1241

1242 self.subfile_expected = [

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

1244 "arguments=foo\n",

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

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

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

1248 "queue\n",

1249 ]

1250

1251 def tearDown(self):

1252 pass

1253

1254 def testWriteWithDagConfig(self):

1255 with temporaryDirectory() as tmp_dir:

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

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

1258 wms_config_filename = "dagman.conf"

1259 wms_configurator = dagman_configurator.DagmanConfigurator(config)

1260 wms_configurator.prepare(wms_config_filename, prefix=tmp_dir)

1261 wms_configurator.configure(self.dag)

1262 dagfile_expected = [

1263 f"CONFIG {wms_config_filename}\n",

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

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

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

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

1268 ]

1269

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

1271

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

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

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

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

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

1277 dagfile_actual = f.readlines()

1278 self.assertEqual(dagfile_actual, dagfile_expected)

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

1280 subfile_actual = f.readlines()

1281 self.assertEqual(subfile_actual, self.subfile_expected)

1282

1283 def testWriteWithoutDagConfig(self):

1284 with temporaryDirectory() as tmp_dir:

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

1286 dagfile_expected = [

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

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

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

1290 ]

1291

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

1293

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

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

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

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

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

1299 dagfile_actual = f.readlines()

1300 self.assertEqual(dagfile_actual, dagfile_expected)

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

1302 subfile_actual = f.readlines()

1303 self.assertEqual(subfile_actual, self.subfile_expected)

1304

1305

1306if __name__ == "__main__":

1307 unittest.main()

Coverage for tests / test_lssthtc.py: 19%

558 statements