Coverage for tests / test_dagman_configurator.py: 30%

1# This file is part of ctrl_bps_htcondor. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <https://www.gnu.org/licenses/>. 

27 

28"""Unit tests for DagmanConfigurator class.""" 

29 

30import logging 

31import os 

32import tempfile 

33import unittest 

34from pathlib import Path 

35 

36from pydantic import ValidationError 

37 

38from lsst.ctrl.bps import BpsConfig 

39from lsst.ctrl.bps.htcondor import HTCDag 

40from lsst.ctrl.bps.htcondor.dagman_configurator import DagmanConfigurator 

41 

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

43 

44 

45class DagmanConfiguratorTestCase(unittest.TestCase): 

46 """Unit tests for DagmanConfigurator class.""" 

47 

48 def setUp(self): 

49 self.config = BpsConfig( 

50 { 

51 "site": { 

52 "foo": { 

53 "wmsConfig": {"DAGMAN_USE_STRICT": 1}, 

54 }, 

55 }, 

56 "wmsConfig": {"DAGMAN_USE_STRICT": 0}, 

57 } 

58 ) 

59 

60 def tearDown(self): 

61 pass 

62 

63 def testInitDefaultSearchOptions(self): 

64 """Test object instantiation with default search options.""" 

65 configurator = DagmanConfigurator(self.config) 

66 self.assertIn("DAGMAN_USE_STRICT", configurator.options) 

67 self.assertEqual(configurator.options["DAGMAN_USE_STRICT"], 0) 

68 self.assertIsNone(configurator.config_path) 

69 self.assertIsNone(configurator.prefix) 

70 

71 def testInitCustomSearchOptions(self): 

72 """Test object instantiation with custom search options.""" 

73 configurator = DagmanConfigurator(self.config, search_opts={"curvals": {"computeSite": "foo"}}) 

74 self.assertIn("DAGMAN_USE_STRICT", configurator.options) 

75 self.assertEqual(configurator.options["DAGMAN_USE_STRICT"], 1) 

76 self.assertIsNone(configurator.config_path) 

77 self.assertIsNone(configurator.prefix) 

78 

79 def testInitWrongOptionType(self): 

80 self.config[".wmsConfig.DAGMAN_USE_STRICT"] = "foo" 

81 with self.assertRaisesRegex(ValidationError, "DAGMAN_USE_STRICT".lower()): 

82 DagmanConfigurator(self.config) 

83 

84 def testInitUnsupportedDagmanOption(self): 

85 """Test object instantiation with unsupported DAGMAN options.""" 

86 self.config[".wmsConfig.DAGMAN_UNSUPPORTED_OPTION"] = "foo" 

87 with self.assertLogs(logger=logger, level="WARNING") as cm: 

88 configurator = DagmanConfigurator(self.config) 

89 self.assertIn("DAGMAN_UNSUPPORTED_OPTION", cm.output[0]) 

90 self.assertNotIn("DAGMAN_UNSUPPORTED_OPTION", configurator.options) 

91 

92 def testInitNoWmsConfig(self): 

93 """Test object instantiation fails when no WMS-specific options.""" 

94 del self.config["wmsConfig"] 

95 with self.assertRaisesRegex(KeyError, "not found"): 

96 DagmanConfigurator(self.config) 

97 

98 def testPrepare(self): 

99 """Test if the method creates the configuration file.""" 

100 configurator = DagmanConfigurator(self.config) 

101 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir: 

102 configurator.prepare("dagman.conf", prefix=tmpdir) 

103 self.assertIn(Path(tmpdir), list(configurator.config_path.parents)) 

104 self.assertTrue(configurator.config_path.is_file()) 

105 self.assertEqual(configurator.config_path.read_text(), "DAGMAN_USE_STRICT = 0") 

106 

107 def testPrepareWithUnsupportedOption(self): 

108 """Test if the method does not include unsupported options.""" 

109 self.config[".wmsConfig.DAGMAN_UNSUPPORTED_OPTION"] = "foo" 

110 configurator = DagmanConfigurator(self.config) 

111 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir: 

112 configurator.prepare("dagman.conf", prefix=tmpdir) 

113 self.assertIn(Path(tmpdir), list(configurator.config_path.parents)) 

114 self.assertTrue(configurator.config_path.is_file()) 

115 self.assertEqual(configurator.config_path.read_text(), "DAGMAN_USE_STRICT = 0") 

116 

117 def testPrepareConfigWriteFailure(self): 

118 """Test if the method raises when it can't create the configuration.""" 

119 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir: 

120 os.chmod(tmpdir, 0o500) 

121 

122 configurator = DagmanConfigurator(self.config) 

123 with self.assertLogs(logger=logger, level="ERROR") as cm, self.assertRaises(OSError): 

124 configurator.prepare("dagman.conf", f"{tmpdir}/subdir") 

125 self.assertIn("Could not write", cm.output[0]) 

126 

127 os.chmod(tmpdir, 0o700) 

128 

129 def testConfigure(self): 

130 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir: 

131 dag = HTCDag(name="test_configure") 

132 configurator = DagmanConfigurator(self.config) 

133 configurator.prepare("dagman.conf", prefix=tmpdir) 

134 configurator.configure(dag) 

135 self.assertIn("bps_wms_config_path", dag.graph["attr"]) 

136 self.assertEqual(dag.graph["attr"]["bps_wms_config_path"], "dagman.conf") 

137 

138 def testConfigureIfNotPrepared(self): 

139 """Test if the method raises when prepare step was skipped.""" 

140 dag = HTCDag(name="test_configure_not_prepared") 

141 configurator = DagmanConfigurator(self.config) 

142 with self.assertRaisesRegex(RuntimeError, "file does not exist"): 

143 configurator.configure(dag)