Coverage for tests / test_bps_utils.py: 22%

1# This file is part of ctrl_bps. 

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

27import logging 

28import shutil 

29import tempfile 

30import unittest 

31from pathlib import Path 

32 

33from lsst.ctrl.bps import BpsConfig 

34from lsst.ctrl.bps.bps_utils import _make_id_link, bps_eval, chdir, mkdir 

35 

36 

37class TestMkdir(unittest.TestCase): 

38 """Test directory creation.""" 

39 

40 def setUp(self): 

41 self.tmpdir = tempfile.TemporaryDirectory() 

42 

43 def tearDown(self): 

44 self.tmpdir.cleanup() 

45 

46 def testSuccess(self): 

47 path = self.tmpdir.name + "/foo/bar" 

48 mkdir(path) 

49 self.assertTrue(Path(path).exists()) 

50 

51 def testFailureDirectoryExists(self): 

52 path = Path(self.tmpdir.name) / "foo/bar" 

53 path.mkdir(parents=True, exist_ok=True) 

54 with self.assertRaisesRegex(OSError, "directory.*exists"): 

55 mkdir(str(path)) 

56 

57 def testFailureOther(self): 

58 # Not checking the error message because it depends on the OS on which 

59 # the test is run ("Permission denied" on linux, "Read-only file 

60 # system" in MacOS). 

61 path = Path("/foo/bar") 

62 with self.assertRaises(OSError): 

63 mkdir(str(path)) 

64 

65 

66class TestChdir(unittest.TestCase): 

67 """Test directory changing.""" 

68 

69 def setUp(self): 

70 self.tmpdir = Path(tempfile.mkdtemp()) 

71 

72 def tearDown(self): 

73 shutil.rmtree(self.tmpdir, ignore_errors=True) 

74 

75 def testSuccessfulChdir(self): 

76 cwd = Path.cwd() 

77 self.assertFalse(self.tmpdir.samefile(cwd)) 

78 with chdir(self.tmpdir): 

79 self.assertTrue(self.tmpdir.samefile(Path.cwd())) 

80 self.assertTrue(cwd.samefile(Path.cwd())) 

81 

82 def testFailingChdir(self): 

83 dir_not_there = self.tmpdir / "notthere" 

84 with self.assertRaises(FileNotFoundError): 

85 with chdir(dir_not_there): 

86 pass # should not get here 

87 

88 

89class TestMakeIdLink(unittest.TestCase): 

90 """Test _make_id_link function.""" 

91 

92 def setUp(self): 

93 self.tmpdir = Path(tempfile.mkdtemp()) 

94 

95 def tearDown(self): 

96 shutil.rmtree(self.tmpdir, ignore_errors=True) 

97 

98 def testMakeIdLinkFalse(self): 

99 """Test skipping making link.""" 

100 config = BpsConfig({"makeIdLink": False}) 

101 submit_path = self.tmpdir / "test_submit/testrun/1" 

102 submit_path.mkdir(parents=True) 

103 config["submitPath"] = str(submit_path) 

104 

105 dir_of_links = self.tmpdir / "bps_links" 

106 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm: 

107 _make_id_link(config, "100.0") 

108 self.assertRegex(cm.records[-1].getMessage(), "Not asked to make id link") 

109 self.assertFalse(dir_of_links.exists()) 

110 

111 def testNoRunID(self): 

112 """Test no link made if no run id.""" 

113 config = BpsConfig({"makeIdLink": True}) 

114 submit_path = self.tmpdir / "test_submit/testrun/2" 

115 submit_path.mkdir(parents=True) 

116 config["submitPath"] = str(submit_path) 

117 

118 dir_of_links = self.tmpdir / "bps_links" 

119 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm: 

120 _make_id_link(config, None) 

121 self.assertRegex(cm.records[-1].getMessage(), "Run ID is None. Skipping making id link.") 

122 self.assertFalse(dir_of_links.exists()) 

123 

124 def testSuccessfulLink(self): 

125 """Test successfully made id link.""" 

126 config = BpsConfig({"makeIdLink": True}) 

127 

128 submit_path = self.tmpdir / "test_submit/testrun/3" 

129 submit_path.mkdir(parents=True) 

130 config["submitPath"] = str(submit_path) 

131 

132 # Make sure can make multiple dirs 

133 dir_of_links = self.tmpdir / "test_bps/links" 

134 config["idLinkPath"] = str(dir_of_links) 

135 

136 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.INFO) as cm: 

137 _make_id_link(config, "100.0") 

138 self.assertRegex(cm.records[-1].getMessage(), "Made id softlink:") 

139 

140 link_path = dir_of_links / "100.0" 

141 self.assertTrue(link_path.is_symlink()) 

142 self.assertEqual(link_path.readlink(), submit_path) 

143 

144 def testSubmitDoesNotExist(self): 

145 """Test checking that submit directory exists.""" 

146 config = BpsConfig({"makeIdLink": True}) 

147 

148 submit_path = self.tmpdir / "test_submit/testrun/4" 

149 submit_path.mkdir(parents=True) 

150 config["submitPath"] = str(submit_path / "notthere") 

151 

152 dir_of_links = self.tmpdir / "test_bps_links" 

153 config["idLinkPath"] = str(dir_of_links) 

154 

155 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.WARNING) as cm: 

156 _make_id_link(config, "100.0") 

157 self.assertRegex( 

158 cm.records[-1].getMessage(), "Could not make id softlink: submitPath does not exist" 

159 ) 

160 self.assertFalse(dir_of_links.exists()) 

161 

162 def testLinkAlreadyExists(self): 

163 """Test skipping if link already correctly exists 

164 for example if a restart gives same id. 

165 """ 

166 config = BpsConfig({"makeIdLink": True}) 

167 

168 submit_path = self.tmpdir / "test_submit/testrun/5" 

169 submit_path.mkdir(parents=True) 

170 config["submitPath"] = str(submit_path) 

171 

172 dir_of_links = self.tmpdir / "test_bps_links" 

173 config["idLinkPath"] = str(dir_of_links) 

174 

175 # Make the softlink 

176 dir_of_links.mkdir(parents=True) 

177 link_path = dir_of_links / "100.0" 

178 link_path.symlink_to(submit_path) 

179 

180 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm: 

181 _make_id_link(config, "100.0") 

182 self.assertRegex(cm.records[-1].getMessage(), "Correct softlink already exists") 

183 self.assertTrue(link_path.is_symlink()) 

184 self.assertEqual(link_path.readlink(), submit_path) 

185 

186 def testFileExistsError(self): 

187 """Test catching of FileExistsError.""" 

188 config = BpsConfig({"makeIdLink": True}) 

189 

190 submit_path = self.tmpdir / "test_submit/testrun/6" 

191 submit_path.mkdir(parents=True) 

192 config["submitPath"] = str(submit_path) 

193 

194 dir_of_links = self.tmpdir / "test_bps_links" 

195 config["idLinkPath"] = str(dir_of_links) 

196 

197 # Make a directory with the link path so 

198 # that get FileExistsError 

199 link_path = dir_of_links / "100.0" 

200 link_path.mkdir(parents=True) 

201 

202 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.WARNING) as cm: 

203 _make_id_link(config, "100.0") 

204 self.assertRegex(cm.records[-1].getMessage(), "Could not make id softlink:.*File exists") 

205 self.assertFalse(link_path.is_symlink()) 

206 

207 def testPermissionError(self): 

208 """Test catching of PermissionError.""" 

209 config = BpsConfig({"makeIdLink": True}) 

210 

211 submit_path = self.tmpdir / "test_submit/testrun/7" 

212 submit_path.mkdir(parents=True) 

213 config["submitPath"] = str(submit_path) 

214 

215 dir_of_links = self.tmpdir / "test_bps_links" 

216 # Create dir without write permissions to cause the error. 

217 dir_of_links.mkdir(mode=0o555, parents=True) 

218 config["idLinkPath"] = str(dir_of_links) 

219 

220 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.WARNING) as cm: 

221 _make_id_link(config, "100.0") 

222 self.assertRegex(cm.records[-1].getMessage(), "Could not make id softlink:.*Permission denied") 

223 link_path = dir_of_links / "100.0" 

224 self.assertFalse(link_path.is_symlink()) 

225 

226 

227class TestBpsEval(unittest.TestCase): 

228 """Test bps_eval function.""" 

229 

230 def testBuiltIn(self): 

231 """Test using a built-in function.""" 

232 with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm: 

233 results = bps_eval("sum", "[1, 2]") 

234 self.assertEqual(results, 3) 

235 self.assertEqual(cm.records[-1].getMessage(), "String passed to eval: 'sum([1, 2])'") 

236 

237 

238if __name__ == "__main__": 

239 unittest.main()