Coverage for tests / test_resource.py: 16%

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 re 

13import unittest 

14 

15from lsst.resources import ResourcePath 

16from lsst.resources.tests import GenericTestCase 

17 

18 

19class ResourceTestCase(GenericTestCase, unittest.TestCase): 

20 """Generic test of resource URIs.""" 

21 

22 scheme = "resource" 

23 netloc = "lsst.resources" 

24 

25 

26class ResourceReadTestCase(unittest.TestCase): 

27 """Test that resource information can be read. 

28 

29 Python package resources are read-only. 

30 """ 

31 

32 # No resources in this package so need a resource in the main 

33 # python distribution. 

34 scheme = "resource" 

35 netloc = "idlelib" 

36 

37 def setUp(self): 

38 self.root = f"{self.scheme}://{self.netloc}" 

39 self.root_uri = ResourcePath(self.root) 

40 

41 def test_read(self): 

42 uri = self.root_uri.join("Icons/README.txt") 

43 self.assertTrue(uri.exists(), f"Check {uri} exists") 

44 

45 content = uri.read().decode() 

46 self.assertIn("IDLE", content) 

47 

48 with uri.as_local() as local_uri: 

49 self.assertEqual(local_uri.scheme, "file") 

50 self.assertTrue(local_uri.exists()) 

51 

52 truncated = uri.read(size=9).decode() 

53 self.assertEqual(truncated, content[:9]) 

54 

55 # Check that directory determination can work directly without the 

56 # trailing slash. 

57 d = self.root_uri.join("Icons") 

58 self.assertTrue(d.isdir()) 

59 self.assertTrue(d.dirLike) 

60 

61 d = self.root_uri.join("Icons/", forceDirectory=True) 

62 self.assertTrue(uri.exists(), f"Check directory {d} exists") 

63 self.assertTrue(d.isdir()) 

64 

65 with self.assertRaises(IsADirectoryError): 

66 with d.as_local() as local_uri: 

67 pass 

68 

69 j = d.join("README.txt") 

70 self.assertEqual(uri, j) 

71 self.assertFalse(j.dirLike) 

72 self.assertFalse(j.isdir()) 

73 not_there = d.join("not-there.yaml") 

74 self.assertFalse(not_there.exists()) 

75 

76 bad = ResourcePath(f"{self.scheme}://bad.module/not.yaml") 

77 multi = ResourcePath.mexists([uri, bad, not_there]) 

78 self.assertTrue(multi[uri]) 

79 self.assertFalse(multi[bad]) 

80 self.assertFalse(multi[not_there]) 

81 

82 # Check that the bad URI works as expected. 

83 self.assertFalse(bad.exists()) 

84 self.assertFalse(bad.isdir()) 

85 with self.assertRaises(FileNotFoundError): 

86 bad.read() 

87 with self.assertRaises(FileNotFoundError): 

88 with bad.as_local(): 

89 pass 

90 with self.assertRaises(FileNotFoundError): 

91 with bad.open("r"): 

92 pass 

93 

94 # fsspec is always not implemented. 

95 with self.assertRaises(NotImplementedError): 

96 bad.to_fsspec() 

97 

98 def test_open(self): 

99 uri = self.root_uri.join("Icons/README.txt") 

100 with uri.open("rb") as buffer: 

101 content = buffer.read() 

102 self.assertEqual(uri.read(), content) 

103 

104 with uri.open("r") as buffer: 

105 content = buffer.read() 

106 self.assertEqual(uri.read().decode(), content) 

107 

108 # Read only. 

109 with self.assertRaises(RuntimeError): 

110 with uri.open("w") as buffer: 

111 pass 

112 

113 def test_walk(self): 

114 """Test that we can find file resources. 

115 

116 Try to find resources in this package. Python does not care whether 

117 a resource is a Python file or anything else. 

118 """ 

119 resource = ResourcePath("resource://lsst.resources/") 

120 resources = set(ResourcePath.findFileResources([resource])) 

121 

122 # Do not try to list all possible options. Files can move around 

123 # and cache files can appear. 

124 subset = { 

125 ResourcePath("resource://lsst.resources/_resourceHandles/_s3ResourceHandle.py"), 

126 ResourcePath("resource://lsst.resources/http.py"), 

127 } 

128 for r in subset: 

129 self.assertIn(r, resources) 

130 

131 resources = set( 

132 ResourcePath.findFileResources( 

133 [ResourcePath("resource://lsst.resources/")], file_filter=r".*\.txt" 

134 ) 

135 ) 

136 self.assertEqual(resources, set()) 

137 

138 # Compare regex with str. 

139 regex = r".*\.py" 

140 py_files_str = list(resource.walk(file_filter=regex)) 

141 py_files_re = list(resource.walk(file_filter=re.compile(regex))) 

142 self.assertGreater(len(py_files_str), 1) 

143 self.assertEqual(py_files_str, py_files_re) 

144 

145 with self.assertRaises(ValueError): 

146 list(ResourcePath("resource://lsst.resources/http.py").walk()) 

147 

148 bad_dir = ResourcePath(f"{self.scheme}://bad.module/a/dir/") 

149 self.assertTrue(bad_dir.isdir()) 

150 with self.assertRaises(ValueError): 

151 list(bad_dir.walk()) 

152 

153 

154if __name__ == "__main__": 

155 unittest.main()