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 }

728

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

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

731VARS job1 num="8"

732VARS job1 spaces="a space"

733PRE_SKIP job1 1

734RETRY job1 3 UNLESS-EXIT 1

735ABORT-DAG-ON job1 100 RETURN 4

736"""

737 mockfh = io.StringIO()

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

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

740

741 def testPartialCommands(self):

742 # Trigger skipping the inner if clauses.

743 dag_cmds = {

744 "pre": {

745 "executable": "exec1",

746 },

747 "post": {

748 "executable": "exec2",

749 },

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

751 "pre_skip": "1",

752 "retry": 3,

753 }

754

755 truth = """SCRIPT PRE job1 exec1

756SCRIPT POST job1 exec2

757VARS job1 num="8"

758VARS job1 spaces="a space"

759PRE_SKIP job1 1

760RETRY job1 3

761"""

762 mockfh = io.StringIO()

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

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

765

766 def testNoCommands(self):

767 dag_cmds = {}

768 mockfh = io.StringIO()

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

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

771

772

773class HTCBackupFilesSinglePathTestCase(unittest.TestCase):

774 """Test htc_backup_files_single_path function."""

775

776 def testSrcDestSame(self):

777 with temporaryDirectory() as tmp_dir:

778 with self.assertRaisesRegex(

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

780 ):

781 lssthtc.htc_backup_files_single_path(tmp_dir, tmp_dir)

782

783 def testSuccess(self):

784 with temporaryDirectory() as tmp_dir:

785 test_tmp_dir = pathlib.Path(tmp_dir)

786 submit_dir = test_tmp_dir / "the_src_dir"

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

788 backup_dir = test_tmp_dir / "the_dest_dir"

789 backup_dir.mkdir()

790 lssthtc.htc_backup_files_single_path(submit_dir, backup_dir)

791 result_submit = []

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

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

794 self.assertEqual(

795 set(result_submit),

796 {

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

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

799 "./tiny_success.dag",

800 },

801 )

802 result_backup = []

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

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

805 self.assertEqual(

806 set(result_backup),

807 {

808 "./tiny_success.info.json",

809 "./tiny_success.dag.metrics",

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

811 "./tiny_success.node_status",

812 },

813 )

814

815

816class HTCBackupFilesTestCase(unittest.TestCase):

817 """Test htc_backup_files function."""

818

819 def testDirectoryNotFound(self):

820 with temporaryDirectory() as tmp_dir:

821 test_tmp_dir = pathlib.Path(tmp_dir)

822 submit_dir = test_tmp_dir / "submit"

823 with self.assertRaises(FileNotFoundError):

824 _ = lssthtc.htc_backup_files(submit_dir)

825

826 def testSuccess(self):

827 with temporaryDirectory() as tmp_dir:

828 test_tmp_dir = pathlib.Path(tmp_dir)

829 submit_dir = test_tmp_dir / "submit"

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

831 result_rescue = lssthtc.htc_backup_files(submit_dir)

832 self.assertIsNone(result_rescue)

833 result_submit = []

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

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

836 self.assertEqual(

837 set(result_submit),

838 {

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

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

841 "./tiny_success.dag",

842 "000/tiny_success.info.json",

843 "000/tiny_success.dag.metrics",

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

845 "000/tiny_success.node_status",

846 },

847 )

848

849 def testDestNotInSubmitDir(self):

850 with temporaryDirectory() as tmp_dir:

851 test_tmp_dir = pathlib.Path(tmp_dir)

852 submit_dir = test_tmp_dir / "submit"

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

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

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

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

857 result_rescue = lssthtc.htc_backup_files(submit_dir)

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

859 result_submit = []

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

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

862 self.assertEqual(

863 set(result_submit),

864 {

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

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

867 "./tiny_problems.dag",

868 "./tiny_problems.dag.rescue001",

869 "001/tiny_problems.info.json",

870 "001/tiny_problems.dag.metrics",

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

872 "001/tiny_problems.node_status",

873 },

874 )

875

876 def testDestInSubmitDir(self):

877 with temporaryDirectory() as tmp_dir:

878 test_tmp_dir = pathlib.Path(tmp_dir)

879 submit_dir = test_tmp_dir / "submit"

880 backup_dir = submit_dir / "subdir"

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

882 result_rescue = lssthtc.htc_backup_files(submit_dir, backup_dir)

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

884 result_submit = []

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

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

887 self.assertEqual(

888 set(result_submit),

889 {

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

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

892 "./tiny_problems.dag",

893 "./tiny_problems.dag.rescue001",

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

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

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

897 "subdir/001/tiny_problems.node_status",

898 },

899 )

900

901 def testRelativeSubdir(self):

902 with temporaryDirectory() as tmp_dir:

903 test_tmp_dir = pathlib.Path(tmp_dir)

904 submit_dir = test_tmp_dir / "submit"

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

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

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

908 result_submit = []

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

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

911 self.assertEqual(

912 set(result_submit),

913 {

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

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

916 "./tiny_problems.dag",

917 "./tiny_problems.dag.rescue001",

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

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

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

921 "reldir/001/tiny_problems.node_status",

922 },

923 )

924

925 def testSubdags(self):

926 with temporaryDirectory() as tmp_dir:

927 test_tmp_dir = pathlib.Path(tmp_dir)

928 submit_dir = test_tmp_dir / "submit"

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

930 result_rescue = lssthtc.htc_backup_files(submit_dir)

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

932 result_submit = []

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

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

935 self.assertEqual(

936 set(result_submit),

937 {

938 "./group_failed_1.dag",

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

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

941 "./group_failed_1.dag.rescue001",

942 "subdags/wms_group_order1_val1a/group_order1_val1a.dag",

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

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

945 "subdags/wms_group_order1_val1b/group_order1_val1b.dag",

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

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

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

949 "subdags/wms_group_order1_val1c/group_order1_val1c.dag",

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

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

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

953 "001/group_failed_1.info.json",

954 "001/group_failed_1.node_status",

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

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

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

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

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

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

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

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

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

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

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

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

967 },

968 )

969

970

971class UpdateRescueFileTestCase(unittest.TestCase):

972 """Test _update_rescue_file function."""

973

974 def testSuccess(self):

975 self.maxDiff = None

976 with temporaryDirectory() as tmp_dir:

977 test_tmp_dir = pathlib.Path(tmp_dir)

978 submit_dir = test_tmp_dir / "submit"

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

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

981 lssthtc._update_rescue_file(rescue_file)

982

983 with open(rescue_file) as fh:

984 lines = fh.readlines()

985 results = "".join(lines)

986

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

988# the u_testuser_DM-46294_group_fail_20250310T160455Z.dag DAG file

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

990# Rescue DAG version: 2.0.1 (partial)

991#

992# Total number of Nodes: 26

993# Nodes premarked DONE: 21

994# Nodes that failed: 2

995# wms_group_order1_val1b,finalJob,<ENDLIST>

996

997DONE pipetaskInit

998DONE label1_val1c_val2a

999DONE label1_val1b_val2b

1000DONE label1_val1b_val2a

1001DONE label1_val1c_val2b

1002DONE label1_val1a_val2a

1003DONE label1_val1a_val2b

1004DONE label3_val1c_val2a

1005DONE label3_val1b_val2b

1006DONE label3_val1b_val2a

1007DONE label3_val1c_val2b

1008DONE label3_val1a_val2a

1009DONE label3_val1a_val2b

1010DONE wms_group_order1_val1a

1011DONE label5_val1a_val2a

1012DONE label5_val1a_val2b

1013DONE wms_group_order1_val1c

1014DONE label5_val1c_val2a

1015DONE label5_val1c_val2b

1016DONE wms_check_status_wms_group_order1_val1a

1017DONE wms_check_status_wms_group_order1_val1c

1018"""

1019

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

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

1022 self.assertEqual(results, truth)

1023

1024

1025class ReadDagStatusTestCase(unittest.TestCase):

1026 """Test read_dag_status function and read_single_dag_status."""

1027

1028 def testFileMissing(self):

1029 with temporaryDirectory() as tmp_dir:

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

1031 _ = lssthtc.read_dag_status(tmp_dir)

1032

1033 def testRegular(self):

1034 with temporaryDirectory() as tmp_dir:

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

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

1037 results = lssthtc.read_dag_status(submit_dir)

1038 truth = {

1039 "JobProcsHeld": 0,

1040 "NodesPost": 0,

1041 "JobProcsIdle": 0,

1042 "NodesTotal": 6,

1043 "NodesFailed": 2,

1044 "NodesDone": 3,

1045 "NodesQueued": 0,

1046 "NodesPre": 0,

1047 "NodesFutile": 1,

1048 "NodesUnready": 0,

1049 }

1050 self.assertEqual(results, results | truth)

1051

1052 def testSubdags(self):

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

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

1055 """

1056 self.maxDiff = None

1057 with temporaryDirectory() as tmp_dir:

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

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

1060 results = lssthtc.read_dag_status(submit_dir)

1061 truth = {

1062 "JobProcsHeld": 0,

1063 "NodesPost": 0,

1064 "JobProcsIdle": 0,

1065 "NodesTotal": 34,

1066 "NodesFailed": 0,

1067 "NodesDone": 17,

1068 "NodesQueued": 3,

1069 "NodesPre": 0,

1070 "NodesFutile": 0,

1071 "NodesUnready": 14,

1072 }

1073 self.assertEqual(results, results | truth)

1074

1075

1076class ReadDagInfoTestCase(unittest.TestCase):

1077 """Test read_dag_info function."""

1078

1079 def testFileMissing(self):

1080 with temporaryDirectory() as tmp_dir:

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

1082 _ = lssthtc.read_dag_info(tmp_dir)

1083

1084 def testSuccess(self):

1085 with temporaryDirectory() as tmp_dir:

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

1087 results = lssthtc.read_dag_info(tmp_dir)

1088

1089 truth = {

1090 "test02": {

1091 "9208.0": {

1092 "ClusterId": 9208,

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

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

1095 "bps_project": "dev",

1096 "bps_payload": "tiny",

1097 "bps_operator": "testuser",

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

1099 "bps_provisioning_job": "provisioningJob",

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

1101 "bps_campaign": "quick",

1102 "bps_runsite": "testpool",

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

1104 "bps_run": "u_testuser_tiny_20250213T164427Z",

1105 "bps_isjob": "True",

1106 }

1107 }

1108 }

1109

1110 self.assertEqual(results, truth)

1111

1112 def testPermissionError(self):

1113 with temporaryDirectory() as tmp_dir:

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

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

1116 mocked_open.side_effect = PermissionError

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

1118 results = lssthtc.read_dag_info(tmp_dir)

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

1120 self.assertEqual({}, results)

1121

1122

1123class HtcWriteCondorFileTestCase(unittest.TestCase):

1124 """Test htc_write_condor_file function."""

1125

1126 def testSuccess(self):

1127 with temporaryDirectory() as tmp_dir:

1128 job_name = "job1"

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

1130 job = {

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

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

1133 "request_memory": "2000",

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

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

1136 }

1137 job_attrs = {

1138 "bps_job_name": job_name,

1139 "bps_job_label": "label1",

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

1141 }

1142 expected = [

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

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

1145 "request_memory=2000\n",

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

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

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

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

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

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

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

1153 "queue\n",

1154 ]

1155

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

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

1158 actual = f.readlines()

1159

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

1161 self.assertTrue(filename.exists())

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

1163 _ = lssthtc.htc_create_submit_from_file(filename)

1164

1165

1166class HtcCreateSubmitFromDagTestCase(unittest.TestCase):

1167 """Test htc_create_submit_from_dag function."""

1168

1169 @classmethod

1170 def setUpClass(cls):

1171 cls.bindir = None

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

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

1174 cls.bindir = tempfile.TemporaryDirectory()

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

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

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

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

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

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

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

1182

1183 @classmethod

1184 def tearDownClass(cls):

1185 if cls.bindir:

1186 cls.bindir.cleanup()

1187

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

1189 def testMaxIdleEnvVar(self):

1190 with temporaryDirectory() as tmp_dir:

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

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

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

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

1195

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

1197 def testMaxIdleGiven(self):

1198 with temporaryDirectory() as tmp_dir:

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

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

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

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

1203

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

1205 def testNoMaxJobsIdle(self):

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

1207 HTCondor version when no MaxIdle passed to from_dag, not

1208 checking arguments string here. Instead just making sure

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

1210 """

1211 with temporaryDirectory() as tmp_dir:

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

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

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

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

1216 mock_param.__contains__.return_value = False

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

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

1219

1220

1221class HtcDagTestCase(unittest.TestCase):

1222 """Test for HTCDag class."""

1223

1224 def setUp(self):

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

1226 job.add_job_cmds(

1227 {

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

1229 "arguments": "foo",

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

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

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

1233 }

1234 )

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

1236

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

1238 self.dag.add_job(job)

1239

1240 self.subfile_expected = [

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

1242 "arguments=foo\n",

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

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

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

1246 "queue\n",

1247 ]

1248

1249 def tearDown(self):

1250 pass

1251

1252 def testWriteWithDagConfig(self):

1253 with temporaryDirectory() as tmp_dir:

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

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

1256 wms_config_filename = "dagman.conf"

1257 wms_configurator = dagman_configurator.DagmanConfigurator(config)

1258 wms_configurator.prepare(wms_config_filename, prefix=tmp_dir)

1259 wms_configurator.configure(self.dag)

1260 dagfile_expected = [

1261 f"CONFIG {wms_config_filename}\n",

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

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

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

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

1266 ]

1267

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

1269

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

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

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

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

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

1275 dagfile_actual = f.readlines()

1276 self.assertEqual(dagfile_actual, dagfile_expected)

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

1278 subfile_actual = f.readlines()

1279 self.assertEqual(subfile_actual, self.subfile_expected)

1280

1281 def testWriteWithoutDagConfig(self):

1282 with temporaryDirectory() as tmp_dir:

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

1284 dagfile_expected = [

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

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

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

1288 ]

1289

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

1291

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

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

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

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

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

1297 dagfile_actual = f.readlines()

1298 self.assertEqual(dagfile_actual, dagfile_expected)

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

1300 subfile_actual = f.readlines()

1301 self.assertEqual(subfile_actual, self.subfile_expected)

1302

1303

1304if __name__ == "__main__":

1305 unittest.main()

Coverage for tests / test_lssthtc.py: 19%

558 statements