Coverage for tests/test_file.py: 26%

1# This file is part of lsst-resources. 

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# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12import contextlib 

13import os 

14import pathlib 

15import unittest 

16import unittest.mock 

17import urllib.parse 

18 

19from lsst.resources import ResourcePath, ResourcePathExpression 

20from lsst.resources.tests import GenericReadWriteTestCase, GenericTestCase 

21 

22TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

23 

24 

25class SimpleTestCase(unittest.TestCase): 

26 """Basic tests for file URIs.""" 

27 

28 def test_instance(self): 

29 for example in ( 

30 "xxx", 

31 ResourcePath("xxx"), 

32 pathlib.Path("xxx"), 

33 urllib.parse.urlparse("file:///xxx"), 

34 ): 

35 self.assertIsInstance(example, ResourcePathExpression) 

36 

37 for example in ({1, 2, 3}, 42, self): 

38 self.assertNotIsInstance(example, ResourcePathExpression) 

39 

40 

41class FileTestCase(GenericTestCase, unittest.TestCase): 

42 """File-specific generic test cases.""" 

43 

44 scheme = "file" 

45 netloc = "localhost" 

46 

47 def test_env_var(self): 

48 """Test that environment variables are expanded.""" 

49 with unittest.mock.patch.dict(os.environ, {"MY_TEST_DIR": "/a/b/c"}): 

50 uri = ResourcePath("${MY_TEST_DIR}/d.txt") 

51 self.assertEqual(uri.path, "/a/b/c/d.txt") 

52 self.assertEqual(uri.scheme, "file") 

53 

54 # This will not expand 

55 uri = ResourcePath("${MY_TEST_DIR}/d.txt", forceAbsolute=False) 

56 self.assertEqual(uri.path, "${MY_TEST_DIR}/d.txt") 

57 self.assertFalse(uri.scheme) 

58 

59 def test_ospath(self): 

60 """File URIs have ospath property.""" 

61 file = ResourcePath(self._make_uri("a/test.txt")) 

62 self.assertEqual(file.ospath, "/a/test.txt") 

63 self.assertEqual(file.ospath, file.path) 

64 

65 # A Schemeless URI can take unquoted files but will be quoted 

66 # when it becomes a file URI. 

67 something = "/a#/???.txt" 

68 file = ResourcePath(something, forceAbsolute=True) 

69 self.assertEqual(file.scheme, "file") 

70 self.assertEqual(file.ospath, something, "From URI: {file}") 

71 self.assertNotIn("???", file.path) 

72 

73 def test_path_lib(self): 

74 """File URIs can be created from pathlib.""" 

75 file = ResourcePath(self._make_uri("a/test.txt")) 

76 

77 path_file = pathlib.Path(file.ospath) 

78 from_path = ResourcePath(path_file) 

79 self.assertEqual(from_path.ospath, file.ospath) 

80 

81 def test_schemeless_root(self): 

82 root = ResourcePath(self._make_uri("/root")) 

83 via_root = ResourcePath("b.txt", root=root) 

84 self.assertEqual(via_root.ospath, "/root/b.txt") 

85 

86 

87TEST_UMASK = 0o0333 

88 

89 

90class FileReadWriteTestCase(GenericReadWriteTestCase, unittest.TestCase): 

91 """File tests involving reading and writing of data.""" 

92 

93 scheme = "file" 

94 netloc = "localhost" 

95 testdir = TESTDIR 

96 transfer_modes = ("move", "copy", "link", "hardlink", "symlink", "relsymlink") 

97 

98 def test_transfer_identical(self): 

99 """Test overwrite of identical files. 

100 

101 Only relevant for local files. 

102 """ 

103 dir1 = self.tmpdir.join("dir1", forceDirectory=True) 

104 dir1.mkdir() 

105 self.assertTrue(dir1.exists()) 

106 dir2 = self.tmpdir.join("dir2", forceDirectory=True) 

107 # A symlink can't include a trailing slash. 

108 dir2_ospath = dir2.ospath 

109 if dir2_ospath.endswith("/"): 

110 dir2_ospath = dir2_ospath[:-1] 

111 os.symlink(dir1.ospath, dir2_ospath) 

112 

113 # Write a test file. 

114 src_file = dir1.join("test.txt") 

115 content = "0123456" 

116 src_file.write(content.encode()) 

117 

118 # Construct URI to destination that should be identical. 

119 dest_file = dir2.join("test.txt") 

120 self.assertTrue(dest_file.exists()) 

121 self.assertNotEqual(src_file, dest_file) 

122 

123 # Transfer it over itself. 

124 dest_file.transfer_from(src_file, transfer="symlink", overwrite=True) 

125 new_content = dest_file.read().decode() 

126 self.assertEqual(content, new_content) 

127 

128 def test_local_temporary(self): 

129 """Create temporary local file if no prefix specified.""" 

130 with ResourcePath.temporary_uri(suffix=".json") as tmp: 

131 self.assertEqual(tmp.getExtension(), ".json", f"uri: {tmp}") 

132 self.assertTrue(tmp.isabs(), f"uri: {tmp}") 

133 self.assertFalse(tmp.exists(), f"uri: {tmp}") 

134 tmp.write(b"abcd") 

135 self.assertTrue(tmp.exists(), f"uri: {tmp}") 

136 self.assertTrue(tmp.isTemporary) 

137 self.assertTrue(tmp.isLocal) 

138 

139 # If we now ask for a local form of this temporary file 

140 # it should still be temporary and it should not be deleted 

141 # on exit. 

142 with tmp.as_local() as loc: 

143 self.assertEqual(tmp, loc) 

144 self.assertTrue(loc.isTemporary) 

145 self.assertTrue(tmp.exists()) 

146 self.assertFalse(tmp.exists(), f"uri: {tmp}") 

147 

148 with ResourcePath.temporary_uri(suffix=".yaml", delete=False) as tmp: 

149 tmp.write(b"1234") 

150 self.assertTrue(tmp.exists(), f"uri: {tmp}") 

151 # If the file doesn't exist there is nothing to clean up so a failure 

152 # here is not a problem. 

153 self.assertTrue(tmp.exists(), f"uri: {tmp} should still exist") 

154 

155 # If removal does not work it's worth reporting that as an error. 

156 tmp.remove() 

157 

158 def test_transfers_from_local(self): 

159 """Extra tests for local transfers.""" 

160 target = self.tmpdir.join("a/target.txt") 

161 with ResourcePath.temporary_uri() as tmp: 

162 tmp.write(b"") 

163 self.assertTrue(tmp.isTemporary) 

164 

165 # Symlink transfers for temporary resources should 

166 # trigger a debug message. 

167 for transfer in ("symlink", "relsymlink"): 

168 with self.assertLogs("lsst.resources", level="DEBUG") as cm: 

169 target.transfer_from(tmp, transfer) 

170 target.remove() 

171 self.assertIn("Using a symlink for a temporary", "".join(cm.output)) 

172 

173 # Force the target directory to be created. 

174 target.transfer_from(tmp, "move") 

175 self.assertFalse(tmp.exists()) 

176 

177 # Temporary file now gone so transfer should not work. 

178 with self.assertRaises(FileNotFoundError): 

179 target.transfer_from(tmp, "move", overwrite=True) 

180 

181 def test_write_with_restrictive_umask(self): 

182 self._test_file_with_restrictive_umask(lambda target: target.write(b"123")) 

183 

184 def test_transfer_from_with_restrictive_umask(self): 

185 def cb(target): 

186 with ResourcePath.temporary_uri() as tmp: 

187 tmp.write(b"") 

188 target.transfer_from(tmp, "copy") 

189 

190 self._test_file_with_restrictive_umask(cb) 

191 

192 def test_mkdir_with_restrictive_umask(self): 

193 self._test_with_restrictive_umask(lambda target: target.mkdir()) 

194 

195 def test_temporary_uri_with_restrictive_umask(self): 

196 with _override_umask(TEST_UMASK): 

197 with ResourcePath.temporary_uri() as tmp: 

198 tmp.write(b"") 

199 self.assertTrue(tmp.exists()) 

200 

201 def _test_file_with_restrictive_umask(self, callback): 

202 def inner_cb(target): 

203 callback(target) 

204 

205 # Make sure the umask was respected for the file itself 

206 file_mode = os.stat(target.ospath).st_mode 

207 self.assertEqual(file_mode & TEST_UMASK, 0) 

208 

209 self._test_with_restrictive_umask(inner_cb) 

210 

211 def _test_with_restrictive_umask(self, callback): 

212 """Make sure that parent directories for a file can be created even if 

213 the user has set a process umask that restricts the write and traverse 

214 bits. 

215 """ 

216 with _override_umask(TEST_UMASK): 

217 target = self.tmpdir.join("a/b/target.txt") 

218 callback(target) 

219 self.assertTrue(target.exists()) 

220 

221 dir_b_path = os.path.dirname(target.ospath) 

222 dir_a_path = os.path.dirname(dir_b_path) 

223 for dir in [dir_a_path, dir_b_path]: 

224 # Make sure we only added the minimum permissions needed for it 

225 # to work (owner-write and owner-traverse) 

226 mode = os.stat(dir).st_mode 

227 self.assertEqual(mode & TEST_UMASK, 0o0300, f"Permissions incorrect for {dir}: {mode:o}") 

228 

229 @unittest.mock.patch("lsst.resources._resourcePath._POOL_EXECUTOR_CLASS", None) 

230 @unittest.mock.patch.dict(os.environ, {"LSST_RESOURCES_EXECUTOR": "process"}) 

231 def test_mexists_process(self) -> None: 

232 """Test mexists with override executor pool. 

233 

234 Force test with process pool. 

235 """ 

236 super().test_mexists() 

237 

238 @unittest.mock.patch("lsst.resources._resourcePath._POOL_EXECUTOR_CLASS", None) 

239 @unittest.mock.patch.dict(os.environ, {"LSST_RESOURCES_EXECUTOR": "process"}) 

240 def test_mtransfer_process(self) -> None: 

241 """Test transfer with override executor pool. 

242 

243 Force test with process pool. 

244 """ 

245 super().test_mtransfer() 

246 

247 

248@contextlib.contextmanager 

249def _override_umask(temp_umask): 

250 old = os.umask(temp_umask) 

251 try: 

252 yield 

253 finally: 

254 os.umask(old) 

255 

256 

257if __name__ == "__main__": 

258 unittest.main()