Coverage for tests / test_provisioner.py: 19%

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"""Unit tests for Provisioner class.""" 

28 

29import logging 

30import os 

31import tempfile 

32import unittest 

33from pathlib import Path 

34 

35from lsst.ctrl.bps import BpsConfig 

36from lsst.ctrl.bps.htcondor import HTCDag 

37from lsst.ctrl.bps.htcondor.provisioner import Provisioner 

38 

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

40 

41 

42class ProvisionerTestCase(unittest.TestCase): 

43 """Unit tests for Provisioner class.""" 

44 

45 def setUp(self): 

46 self.config = BpsConfig({}, defaults={"wmsServiceClass": "lsst.ctrl.bps.htcondor.HTCondorService"}) 

47 

48 def tearDown(self): 

49 pass 

50 

51 def testConfigureWithoutExistingConfig(self): 

52 """Test if the configuration file is created if necessary.""" 

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

54 filename = f"{tmpdir}/condor-info.py" 

55 self.config[".provisioning.provisioningScriptConfig"] = "foo" 

56 self.config[".provisioning.provisioningScriptConfigPath"] = filename 

57 

58 provisioner = Provisioner(self.config) 

59 with self.assertLogs(level="INFO") as cm: 

60 provisioner.configure() 

61 

62 self.assertRegex(cm.output[0], "file.*not found") 

63 self.assertRegex(cm.output[0], "new one") 

64 self.assertTrue(Path(filename).is_file()) 

65 self.assertEqual(Path(filename).read_text(), "foo") 

66 self.assertTrue(provisioner.is_configured) 

67 

68 def testConfigureWithExistingConfig(self): 

69 """Test if the existing configuration file is left unchanged.""" 

70 with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as tmpfile: 

71 self.config[".provisioning.provisioningScriptConfig"] = "bar" 

72 self.config[".provisioning.provisioningScriptConfigPath"] = tmpfile.name 

73 tmpfile.write("foo") 

74 tmpfile.flush() 

75 

76 provisioner = Provisioner(self.config) 

77 with self.assertLogs(level="INFO") as cm: 

78 provisioner.configure() 

79 

80 self.assertRegex(cm.output[0], "file.*exists") 

81 self.assertEqual(Path(tmpfile.name).read_text(), "foo") 

82 self.assertTrue(provisioner.is_configured) 

83 

84 def testConfigureNoConfigRequired(self): 

85 """Test when no configuration file is required.""" 

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

87 filename = f"{tmpdir}/condor-info.py" 

88 self.config[".provisioning.provisioningScriptConfig"] = "" 

89 self.config[".provisioning.provisioningScriptConfigPath"] = filename 

90 

91 provisioner = Provisioner(self.config) 

92 with self.assertLogs(level="INFO") as cm: 

93 provisioner.configure() 

94 

95 self.assertRegex(cm.output[0], "Configuration.*not provided") 

96 self.assertFalse(Path(filename).is_file()) 

97 self.assertTrue(provisioner.is_configured) 

98 

99 def testConfigureOsError(self): 

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

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

102 filename = f"{tmpdir}/subdir/condor-info.py" 

103 self.config[".provisioning.provisioningScriptConfigPath"] = filename 

104 os.chmod(tmpdir, 0o500) 

105 

106 provisioner = Provisioner(self.config) 

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

108 provisioner.configure() 

109 

110 self.assertRegex(cm.output[0], "Cannot create configuration") 

111 

112 os.chmod(tmpdir, 0o700) 

113 

114 def testPrepare(self): 

115 """Test if the provisioning script is created.""" 

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

117 prov_config = Path(f"{tmpdir}/condor-info.py") 

118 self.config[".provisioning.provisioningScriptConfigPath"] = str(prov_config) 

119 self.config[".provisioning.provisioningScriptConfig"] = "foo" 

120 self.config[".provisioning.provisioningScript"] = "bar" 

121 

122 prov_script = Path(tmpdir) / "provisioning_job.sh" 

123 provisioner = Provisioner(self.config) 

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

125 provisioner.configure() 

126 provisioner.prepare(prov_script.name, prefix=prov_script.parent) 

127 

128 self.assertRegex(cm.output[1], "Writing.*provisioning script") 

129 self.assertTrue(provisioner.is_configured) 

130 self.assertTrue(prov_config.is_file()) 

131 self.assertEqual(prov_config.read_text(), "foo") 

132 self.assertTrue(prov_script.is_file()) 

133 self.assertEqual(prov_script.read_text(), "bar") 

134 

135 def testPrepareIfNotConfigured(self): 

136 """Test if the method raises when the configuration step is skipped.""" 

137 provisioner = Provisioner(self.config) 

138 with self.assertRaises(RuntimeError): 

139 provisioner.prepare("provisioning_job.sh", prefix="") 

140 

141 def testProvision(self): 

142 """Test if the provisioning job is added to the DAG.""" 

143 script = Path("provisioningJob.sh") 

144 cmds = { 

145 "universe": "local", 

146 "executable": str(script), 

147 "should_transfer_files": "NO", 

148 "getenv": "True", 

149 "output": f"jobs/{script.stem}/{script.stem}.$(Cluster).out", 

150 "error": f"jobs/{script.stem}/{script.stem}.$(Cluster).out", 

151 "log": f"jobs/{script.stem}/{script.stem}.$(Cluster).log", 

152 } 

153 dag = HTCDag(name="default") 

154 

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

156 self.config[".provisioning.provisioningMaxWallTime"] = 3600 

157 self.config[".provisioning.provisioningScriptConfigPath"] = f"{tmpdir}/condor-info.py" 

158 self.config[".bps_defined.nodeset"] = "20260129T163505Z" 

159 

160 provisioner = Provisioner(self.config) 

161 provisioner.configure() 

162 provisioner.prepare(script.name, prefix=tmpdir) 

163 provisioner.provision(dag) 

164 

165 self.assertIsNotNone(dag.graph["service_job"]) 

166 self.assertEqual(dag.graph["service_job"].name, script.stem) 

167 self.assertEqual(dag.graph["service_job"].label, script.stem) 

168 self.assertEqual(dict(dag.graph["service_job"].cmds), cmds) 

169 self.assertIn("bps_provisioning_job", dag.graph["attr"]) 

170 

171 def testProvisionError(self): 

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

173 dag = HTCDag(name="test") 

174 provisioner = Provisioner(self.config) 

175 with self.assertRaises(RuntimeError): 

176 provisioner.provision(dag) 

177 

178 

179if __name__ == "__main__": 

180 unittest.main()