Coverage for tests / test_eups.py: 29%
155 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 08:38 +0000
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.
12import os
13import re
14import sys
15import unittest
16import unittest.mock
18from lsst.resources import ResourceInfo, ResourcePath
19from lsst.resources.eups import EupsResourcePath
20from lsst.resources.tests import GenericTestCase
22TESTDIR = os.path.abspath(os.path.dirname(__file__))
23PKG_DIR = os.path.join(TESTDIR, "packages", "eups")
24PACKAGE_PATH = os.path.join(TESTDIR, "packages")
27class EupsTestCase(GenericTestCase, unittest.TestCase):
28 """Generic test of resource URIs."""
30 scheme = "eups"
31 netloc = "pkg"
33 @classmethod
34 def setUpClass(cls) -> None:
35 # The actual value does not matter for these tests.
36 os.environ["PKG_DIR"] = PKG_DIR
37 super().setUpClass()
39 @classmethod
40 def tearDownClass(cls) -> None:
41 del os.environ["PKG_DIR"]
42 super().tearDownClass()
44 def test_relative(self) -> None:
45 # This test uses two additional netlocs which need corresponding
46 # environment variables to function. The values do not matter.
47 with unittest.mock.patch.dict(os.environ, {"OTHER_DIR": "x", "MY.HOST_DIR": "y"}):
48 super().test_relative()
51class EupsReadTestCase(unittest.TestCase):
52 """Test that EUPS information can be read.
54 EUPS resources can be thought of as being read only even if the
55 underlying URI is a ``file`` URI.
56 """
58 scheme = "eups"
59 netloc = "pkg"
61 @classmethod
62 def setUpClass(cls) -> None:
63 # The actual value does not matter for these tests.
64 os.environ["PKG_DIR"] = PKG_DIR
65 super().setUpClass()
67 @classmethod
68 def tearDownClass(cls) -> None:
69 del os.environ["PKG_DIR"]
70 super().tearDownClass()
72 def setUp(self):
73 self.root = f"{self.scheme}://{self.netloc}"
74 self.root_uri = ResourcePath(self.root)
76 def test_read(self):
77 uri = self.root_uri.join("config/test.txt")
78 self.assertTrue(uri.exists(), f"Check {uri} exists")
80 content = uri.read().decode()
81 self.assertIn("A test config.", content)
83 with uri.as_local() as local_uri:
84 self.assertEqual(local_uri.scheme, "file")
85 self.assertTrue(local_uri.exists())
87 truncated = uri.read(size=9).decode()
88 self.assertEqual(truncated, content[:9])
90 # Check that directory determination can work directly without the
91 # trailing slash.
92 d = self.root_uri.join("config")
93 self.assertTrue(d.isdir())
94 self.assertTrue(d.dirLike)
96 d = self.root_uri.join("config/", forceDirectory=True)
97 self.assertTrue(d._proxy.dirLike) # Ensure that dir-ness propagates to proxy.
98 self.assertTrue(d.exists(), f"Check directory {d} exists")
99 self.assertTrue(d.isdir())
101 with self.assertRaises(IsADirectoryError):
102 with d.as_local() as local_uri:
103 pass
105 j = d.join("test.txt")
106 self.assertEqual(uri, j)
107 self.assertFalse(j.dirLike)
108 self.assertFalse(j.isdir())
109 not_there = d.join("not-there.yaml")
110 self.assertFalse(not_there.exists())
112 bad = ResourcePath(f"{self.scheme}://bad.module/not.yaml")
113 multi = ResourcePath.mexists([uri, bad, not_there])
114 self.assertTrue(multi[uri])
115 self.assertFalse(multi[bad])
116 self.assertFalse(multi[not_there])
118 # Check that the bad URI works as expected.
119 self.assertFalse(bad.exists())
120 self.assertFalse(bad.isdir())
121 with self.assertRaises(FileNotFoundError):
122 bad.read()
123 with self.assertRaises(FileNotFoundError):
124 with bad.as_local():
125 pass
126 with self.assertRaises(FileNotFoundError):
127 with bad.open("r"):
128 pass
130 # fsspec is always not implemented.
131 with self.assertRaises(NotImplementedError):
132 bad.to_fsspec()
134 def test_open(self):
135 uri = self.root_uri.join("config/test.txt")
136 with uri.open("rb") as buffer:
137 content = buffer.read()
138 self.assertEqual(uri.read(), content)
140 with uri.open("r") as buffer:
141 content = buffer.read()
142 self.assertEqual(uri.read().decode(), content)
144 def test_get_info(self):
145 file_uri = self.root_uri.join("config/test.txt")
146 info = file_uri.get_info()
147 self.assertIsInstance(info, ResourceInfo)
148 self.assertEqual(info.uri, str(file_uri))
149 self.assertTrue(info.is_file)
150 self.assertGreater(info.size, 0)
152 dir_uri = self.root_uri.join("config/", forceDirectory=True)
153 dirinfo = dir_uri.get_info()
154 self.assertIsInstance(dirinfo, ResourceInfo)
155 self.assertEqual(dirinfo.uri, str(dir_uri))
156 self.assertFalse(dirinfo.is_file)
157 self.assertEqual(dirinfo.size, 0)
159 def test_walk(self):
160 """Test that we can find file resources.
162 Try to find resources in this package. Python does not care whether
163 a resource is a Python file or anything else.
164 """
165 resource = ResourcePath(f"{self.scheme}://{self.netloc}/")
166 resources = set(ResourcePath.findFileResources([resource]))
168 # Make sure that walk() can run and knows this is a directory.
169 cfg = resource.join("config/")
170 _, dirs, files = next(cfg.walk())
171 self.assertEqual(dirs, [])
172 self.assertEqual(set(files), {"test.txt", "test2.yaml"})
174 # Do not try to list all possible options. Files can move around
175 # and cache files can appear.
176 subset = {
177 ResourcePath(f"{self.scheme}://{self.netloc}/config/test.txt"),
178 ResourcePath(f"{self.scheme}://{self.netloc}/config/test2.yaml"),
179 }
180 for r in subset:
181 self.assertIn(r, resources)
183 resources = set(
184 ResourcePath.findFileResources(
185 [ResourcePath(f"{self.scheme}://{self.netloc}/")], file_filter=r".*\.json"
186 )
187 )
188 self.assertEqual(resources, set())
190 # Compare regex with str.
191 regex = r".*\.yaml"
192 y_files_str = list(resource.walk(file_filter=regex))
193 y_files_re = list(resource.walk(file_filter=re.compile(regex)))
194 self.assertGreater(len(y_files_str), 1)
195 self.assertEqual(y_files_str, y_files_re)
197 bad_dir = ResourcePath(f"{self.scheme}://bad.module/a/dir/")
198 self.assertTrue(bad_dir.isdir())
199 with self.assertRaises(ValueError):
200 list(bad_dir.walk())
202 def test_env_var(self):
203 """Test that environment variables are converted."""
204 with unittest.mock.patch.dict(os.environ, {"MY_TEST_DIR": TESTDIR}):
205 for env_string in ("$MY_TEST_DIR", "${MY_TEST_DIR}"):
206 uri = ResourcePath(f"{env_string}/data/dir1/a.yaml")
207 self.assertEqual(uri.path, "/data/dir1/a.yaml")
208 self.assertEqual(uri.scheme, "eups")
209 self.assertEqual(uri.netloc, "my_test")
210 self.assertTrue(uri.exists())
213class EupsAsResourcesReadTestCase(EupsReadTestCase):
214 """Test that EUPS information can be read via resources.
216 EUPS resources can be thought of as being read only even if the
217 underlying URI is a ``file`` URI.
218 """
220 scheme = "eups"
221 netloc = "pkg1"
223 @classmethod
224 def setUpClass(cls) -> None:
225 # The actual value does not matter for these tests.
226 sys.path.append(PACKAGE_PATH)
227 super().setUpClass()
229 @classmethod
230 def tearDownClass(cls) -> None:
231 sys.path.remove(PACKAGE_PATH)
232 super().tearDownClass()
235class EupsAsResourcesReadTestCase2(EupsAsResourcesReadTestCase):
236 """Test that EUPS information can be read via resources with lsst-style
237 package.
239 EUPS resources can be thought of as being read only even if the
240 underlying URI is a ``file`` URI.
241 """
243 scheme = "eups"
244 netloc = "pkg2_sub"
246 @classmethod
247 def setUpClass(cls) -> None:
248 # To avoid confusion with other lsst packages, override the default
249 # prefix that is added to the EUPS name.
250 cls.prefix = EupsResourcePath._default_namespace
251 EupsResourcePath._default_namespace = "prefix"
252 super().setUpClass()
254 @classmethod
255 def tearDownClass(cls) -> None:
256 EupsResourcePath._default_namespace = cls.prefix
257 super().tearDownClass()
260if __name__ == "__main__":
261 unittest.main()