Coverage for tests / test_file.py: 25%

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 datetime 

14import os 

15import pathlib 

16import unittest 

17import unittest.mock 

18import urllib.parse 

19 

20from lsst.resources import ResourceInfo, ResourcePath, ResourcePathExpression 

21from lsst.resources.tests import GenericReadWriteTestCase, GenericTestCase 

22 

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

24 

25 

26class SimpleTestCase(unittest.TestCase): 

27 """Basic tests for file URIs.""" 

28 

29 def test_instance(self): 

30 for example in ( 

31 "xxx", 

32 ResourcePath("xxx"), 

33 pathlib.Path("xxx"), 

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

35 ): 

36 self.assertIsInstance(example, ResourcePathExpression) 

37 

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

39 self.assertNotIsInstance(example, ResourcePathExpression) 

40 

41 

42class FileTestCase(GenericTestCase, unittest.TestCase): 

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

44 

45 scheme = "file" 

46 netloc = "localhost" 

47 

48 def test_env_var(self): 

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

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

51 uri = ResourcePath("${MY_TEST_DIRX}/d.txt") 

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

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

54 

55 # This will not expand 

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

57 self.assertEqual(uri.path, "${MY_TEST_DIRX}/d.txt") 

58 self.assertFalse(uri.scheme) 

59 

60 def test_ospath(self): 

61 """File URIs have ospath property.""" 

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

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

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

65 

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

67 # when it becomes a file URI. 

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

69 file = ResourcePath(something, forceAbsolute=True) 

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

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

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

73 

74 def test_path_lib(self): 

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

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

77 

78 path_file = pathlib.Path(file.ospath) 

79 from_path = ResourcePath(path_file) 

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

81 

82 def test_schemeless_root(self): 

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

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

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

86 

87 def test_get_info(self): 

88 now = datetime.datetime.now(tz=datetime.UTC) 

89 with ResourcePath.temporary_uri(suffix=".txt") as target: 

90 target.write(b"abc") 

91 

92 info = target.get_info() 

93 self.assertIsInstance(info, ResourceInfo) 

94 self.assertTrue(info.uri.endswith(".txt")) 

95 self.assertTrue(info.is_file) 

96 self.assertEqual(info.size, 3) 

97 self.assertEqual(info.checksums, {}) 

98 self.assertEqual(info.last_modified.tzinfo, datetime.UTC) 

99 self.assertGreaterEqual(info.last_modified.timestamp(), now.timestamp() - 1.0) 

100 

101 dirinfo = target.parent().get_info() 

102 self.assertEqual(dirinfo.uri, str(target.parent())) 

103 self.assertFalse(dirinfo.is_file) 

104 self.assertEqual(dirinfo.size, 0) 

105 self.assertGreaterEqual(dirinfo.last_modified.timestamp(), 0) 

106 self.assertEqual(dirinfo.checksums, {}) 

107 

108 

109TEST_UMASK = 0o0333 

110 

111 

112class FileReadWriteTestCase(GenericReadWriteTestCase, unittest.TestCase): 

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

114 

115 scheme = "file" 

116 netloc = "localhost" 

117 testdir = TESTDIR 

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

119 

120 def test_transfer_identical(self): 

121 """Test overwrite of identical files. 

122 

123 Only relevant for local files. 

124 """ 

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

126 dir1.mkdir() 

127 self.assertTrue(dir1.exists()) 

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

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

130 dir2_ospath = dir2.ospath 

131 if dir2_ospath.endswith("/"): 

132 dir2_ospath = dir2_ospath[:-1] 

133 os.symlink(dir1.ospath, dir2_ospath) 

134 

135 # Write a test file. 

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

137 content = "0123456" 

138 src_file.write(content.encode()) 

139 

140 # Construct URI to destination that should be identical. 

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

142 self.assertTrue(dest_file.exists()) 

143 self.assertNotEqual(src_file, dest_file) 

144 

145 # Transfer it over itself. 

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

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

148 self.assertEqual(content, new_content) 

149 

150 def test_local_temporary(self): 

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

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

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

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

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

156 tmp.write(b"abcd") 

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

158 self.assertTrue(tmp.isTemporary) 

159 self.assertTrue(tmp.isLocal) 

160 

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

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

163 # on exit. 

164 with tmp.as_local() as loc: 

165 self.assertEqual(tmp, loc) 

166 self.assertTrue(loc.isTemporary) 

167 self.assertTrue(tmp.exists()) 

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

169 

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

171 tmp.write(b"1234") 

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

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

174 # here is not a problem. 

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

176 

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

178 tmp.remove() 

179 

180 def test_transfers_from_local(self): 

181 """Extra tests for local transfers.""" 

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

183 with ResourcePath.temporary_uri() as tmp: 

184 tmp.write(b"") 

185 self.assertTrue(tmp.isTemporary) 

186 

187 # Symlink transfers for temporary resources should 

188 # trigger a debug message. 

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

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

191 target.transfer_from(tmp, transfer) 

192 target.remove() 

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

194 

195 # Force the target directory to be created. 

196 target.transfer_from(tmp, "move") 

197 self.assertFalse(tmp.exists()) 

198 

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

200 with self.assertRaises(FileNotFoundError): 

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

202 

203 def test_write_with_restrictive_umask(self): 

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

205 

206 def test_transfer_from_with_restrictive_umask(self): 

207 def cb(target): 

208 with ResourcePath.temporary_uri() as tmp: 

209 tmp.write(b"") 

210 target.transfer_from(tmp, "copy") 

211 

212 self._test_file_with_restrictive_umask(cb) 

213 

214 def test_mkdir_with_restrictive_umask(self): 

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

216 

217 def test_temporary_uri_with_restrictive_umask(self): 

218 with _override_umask(TEST_UMASK): 

219 with ResourcePath.temporary_uri() as tmp: 

220 tmp.write(b"") 

221 self.assertTrue(tmp.exists()) 

222 

223 def _test_file_with_restrictive_umask(self, callback): 

224 def inner_cb(target): 

225 callback(target) 

226 

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

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

229 self.assertEqual(file_mode & TEST_UMASK, 0) 

230 

231 self._test_with_restrictive_umask(inner_cb) 

232 

233 def _test_with_restrictive_umask(self, callback): 

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

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

236 bits. 

237 """ 

238 with _override_umask(TEST_UMASK): 

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

240 callback(target) 

241 self.assertTrue(target.exists()) 

242 

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

244 dir_a_path = os.path.dirname(dir_b_path) 

245 for dir in [dir_a_path, dir_b_path]: 

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

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

248 mode = os.stat(dir).st_mode 

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

250 

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

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

253 def test_mexists_process(self) -> None: 

254 """Test mexists with override executor pool. 

255 

256 Force test with process pool. 

257 """ 

258 super().test_mexists() 

259 

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

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

262 def test_mtransfer_process(self) -> None: 

263 """Test transfer with override executor pool. 

264 

265 Force test with process pool. 

266 """ 

267 super().test_mtransfer() 

268 

269 

270@contextlib.contextmanager 

271def _override_umask(temp_umask): 

272 old = os.umask(temp_umask) 

273 try: 

274 yield 

275 finally: 

276 os.umask(old) 

277 

278 

279if __name__ == "__main__": 

280 unittest.main()