Coverage for python/lsst/resources/tests.py: 8%
672 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-24 01:20 -0800
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.
11from __future__ import annotations
13__all__ = ["GenericReadWriteTestCase", "GenericTestCase"]
15import logging
16import os
17import pathlib
18import random
19import string
20import sys
21import tempfile
22import unittest
23import urllib.parse
24import uuid
25from collections.abc import Iterable
26from typing import TYPE_CHECKING, Any
28try:
29 import fsspec
30except ImportError:
31 fsspec = None
33from lsst.resources import ResourcePath
34from lsst.resources.utils import makeTestTempDir, removeTestTempDir
36TESTDIR = os.path.abspath(os.path.dirname(__file__))
39def _check_open(
40 test_case: _GenericTestCase | unittest.TestCase,
41 uri: ResourcePath,
42 *,
43 mode_suffixes: Iterable[str] = ("", "t", "b"),
44 **kwargs: Any,
45) -> None:
46 """Test an implementation of ButlerURI.open.
48 Parameters
49 ----------
50 test_case : `unittest.TestCase`
51 Test case to use for assertions.
52 uri : `ResourcePath`
53 URI to use for tests. Must point to a writeable location that is not
54 yet occupied by a file. On return, the location may point to a file
55 only if the test fails.
56 mode_suffixes : `~collections.abc.Iterable` of `str`
57 Suffixes to pass as part of the ``mode`` argument to
58 `ResourcePath.open`, indicating whether to open as binary or as text;
59 the only permitted elements are ``""``, ``"t"``, and ``"b"`.
60 **kwargs
61 Additional keyword arguments to forward to all calls to `open`.
62 """
63 text_content = "abcdefghijklmnopqrstuvwxyz🙂"
64 bytes_content = uuid.uuid4().bytes
65 content_by_mode_suffix: dict[str, str | bytes] = {
66 "": text_content,
67 "t": text_content,
68 "b": bytes_content,
69 }
70 empty_content_by_mode_suffix: dict[str, str | bytes] = {
71 "": "",
72 "t": "",
73 "b": b"",
74 }
75 # To appease mypy
76 double_content_by_mode_suffix: dict[str, str | bytes] = {
77 "": text_content + text_content,
78 "t": text_content + text_content,
79 "b": bytes_content + bytes_content,
80 }
81 for mode_suffix in mode_suffixes:
82 content = content_by_mode_suffix[mode_suffix]
83 double_content = double_content_by_mode_suffix[mode_suffix]
84 # Create file with mode='x', which prohibits overwriting.
85 with uri.open("x" + mode_suffix, **kwargs) as write_buffer:
86 write_buffer.write(content)
87 test_case.assertTrue(uri.exists())
88 # Check that opening with 'x' now raises, and does not modify content.
89 with test_case.assertRaises(FileExistsError):
90 with uri.open("x" + mode_suffix, **kwargs) as write_buffer:
91 write_buffer.write("bad")
92 # Read the file we created and check the contents.
93 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
94 test_case.assertEqual(read_buffer.read(), content)
95 # The names will not match if a local temporary is being written.
96 if not kwargs.get("prefer_file_temporary"):
97 test_case.assertIn(uri.basename(), read_buffer.name)
98 # Check that we can read bytes in a loop and get EOF
99 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
100 # Seek off the end of the file and should read empty back.
101 read_buffer.seek(1024)
102 test_case.assertEqual(read_buffer.tell(), 1024)
103 content_read = read_buffer.read() # Read as much as we can.
104 test_case.assertEqual(len(content_read), 0, f"Read: {content_read!r}, expected empty.")
106 # First read more than the content.
107 read_buffer.seek(0)
108 size = len(content) * 3
109 chunk_read = read_buffer.read(size)
110 test_case.assertEqual(chunk_read, content)
112 # Repeated reads should always return empty string.
113 chunk_read = read_buffer.read(size)
114 test_case.assertEqual(len(chunk_read), 0)
115 chunk_read = read_buffer.read(size)
116 test_case.assertEqual(len(chunk_read), 0)
118 # Go back to start of file and read in smaller chunks.
119 read_buffer.seek(0)
120 size = len(content) // 3
122 content_read = empty_content_by_mode_suffix[mode_suffix]
123 n_reads = 0
124 while chunk_read := read_buffer.read(size):
125 content_read += chunk_read
126 n_reads += 1
127 if n_reads > 10: # In case EOF never hits because of bug.
128 raise AssertionError(
129 f"Failed to stop reading from file after {n_reads} loops. "
130 f"Read {len(content_read)} bytes/characters. Expected {len(content)}."
131 )
132 test_case.assertEqual(content_read, content)
134 # Go back to start of file and read the entire thing.
135 read_buffer.seek(0)
136 content_read = read_buffer.read()
137 test_case.assertEqual(content_read, content)
139 # Seek off the end of the file and should read empty back.
140 # We run this check twice since in some cases the handle will
141 # cache knowledge of the file size.
142 read_buffer.seek(1024)
143 test_case.assertEqual(read_buffer.tell(), 1024)
144 content_read = read_buffer.read()
145 test_case.assertEqual(len(content_read), 0, f"Read: {content_read!r}, expected empty.")
147 # Write multiple chunks with flushing to ensure that any handles that
148 # cache without flushing work properly.
149 n = 3
150 with uri.open("w" + mode_suffix, **kwargs) as write_buffer:
151 for _ in range(n):
152 write_buffer.write(content)
153 write_buffer.flush()
154 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
155 test_case.assertEqual(read_buffer.read(), content * n)
157 # Write two copies of the content, overwriting the single copy there.
158 with uri.open("w" + mode_suffix, **kwargs) as write_buffer:
159 write_buffer.write(double_content)
160 # Read again, this time use mode='r+', which reads what is there and
161 # then lets us write more; we'll use that to reset the file to one
162 # copy of the content.
163 with uri.open("r+" + mode_suffix, **kwargs) as rw_buffer:
164 test_case.assertEqual(rw_buffer.read(), double_content)
165 rw_buffer.seek(0)
166 rw_buffer.truncate()
167 rw_buffer.write(content)
168 rw_buffer.seek(0)
169 test_case.assertEqual(rw_buffer.read(), content)
170 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
171 test_case.assertEqual(read_buffer.read(), content)
172 # Append some more content to the file; should now have two copies.
173 with uri.open("a" + mode_suffix, **kwargs) as append_buffer:
174 append_buffer.write(content)
175 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
176 test_case.assertEqual(read_buffer.read(), double_content)
177 # Final mode to check is w+, which does read/write but truncates first.
178 with uri.open("w+" + mode_suffix, **kwargs) as rw_buffer:
179 test_case.assertEqual(rw_buffer.read(), empty_content_by_mode_suffix[mode_suffix])
180 rw_buffer.write(content)
181 rw_buffer.seek(0)
182 test_case.assertEqual(rw_buffer.read(), content)
183 # Check that two seeks with reads to end return correctly.
184 # Seek is only reliable with "b" mode.
185 if mode_suffix == "b":
186 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
187 size = len(content)
188 seek1 = 2 * size // 3
189 read_buffer.seek(seek1)
190 content1 = read_buffer.read()
191 test_case.assertEqual(len(content1), size - seek1)
192 # Seek earlier and then read to end.
193 seek2 = size // 2
194 read_buffer.seek(seek2)
195 content2 = read_buffer.read()
196 test_case.assertEqual(len(content2), size - seek2)
197 # Check that we can seek from end and read and seek and read.
198 # Negative seek only works in binary mode.
199 with uri.open("rb", **kwargs) as read_buffer:
200 read_buffer.seek(-5, 2) # Relative to end
201 content_read = read_buffer.read()
202 test_case.assertEqual(len(content_read), 5)
203 read_buffer.seek(-10, 2) # Relative to end
204 content_read = read_buffer.read()
205 test_case.assertEqual(len(content_read), 10)
206 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
207 test_case.assertEqual(read_buffer.read(), content)
208 # Remove file to make room for the next loop of tests with this URI.
209 uri.remove()
212if TYPE_CHECKING:
214 class TestCaseMixin(unittest.TestCase):
215 """Base class for mixin test classes that use TestCase methods."""
217 pass
219else:
221 class TestCaseMixin:
222 """Do-nothing definition of mixin base class for regular execution."""
224 pass
227class _GenericTestCase(TestCaseMixin):
228 """Generic base class for test mixin."""
230 scheme: str | None = None
231 netloc: str | None = None
232 base_path: str | None = None
233 path1 = "test_dir"
234 path2 = "file.txt"
236 def _make_uri(self, path: str, netloc: str | None = None) -> str:
237 if self.scheme is not None:
238 if netloc is None:
239 netloc = self.netloc
240 if path.startswith("/"):
241 path = path[1:]
242 if self.base_path is not None:
243 path = f"{self.base_path}/{path}".lstrip("/")
245 return f"{self.scheme}://{netloc}/{path}"
246 else:
247 return path
250class GenericTestCase(_GenericTestCase):
251 """Test cases for generic manipulation of a `ResourcePath`."""
253 def setUp(self) -> None:
254 if self.scheme is None:
255 raise unittest.SkipTest("No scheme defined")
256 self.root = self._make_uri("")
257 self.root_uri = ResourcePath(self.root, forceDirectory=True, forceAbsolute=False)
259 def test_creation(self) -> None:
260 self.assertEqual(self.root_uri.scheme, self.scheme)
261 self.assertEqual(self.root_uri.netloc, self.netloc)
262 self.assertFalse(self.root_uri.query)
263 self.assertFalse(self.root_uri.params)
265 with self.assertRaises(ValueError):
266 ResourcePath({}) # type: ignore
268 with self.assertRaises(RuntimeError):
269 ResourcePath(self.root_uri, isTemporary=True)
271 file = self.root_uri.join("file.txt", forceDirectory=False)
272 with self.assertRaises(RuntimeError):
273 ResourcePath(file, forceDirectory=True)
275 file = self.root_uri.join("file.txt")
276 file_as_dir = ResourcePath(file, forceDirectory=True)
277 self.assertTrue(file_as_dir.isdir())
279 dir = self._make_uri("a/b/c/")
280 with self.assertRaises(ValueError):
281 ResourcePath(dir, forceDirectory=False)
283 with self.assertRaises(NotImplementedError):
284 ResourcePath("unknown://netloc")
286 replaced = file.replace(fragment="frag")
287 self.assertEqual(replaced.fragment, "frag")
289 with self.assertRaises(ValueError):
290 file.replace(scheme="new")
292 self.assertNotEqual(replaced, str(replaced))
293 self.assertNotEqual(str(replaced), replaced)
295 def test_extension(self) -> None:
296 uri = ResourcePath(self._make_uri("dir/test.txt"))
297 self.assertEqual(uri.updatedExtension(None), uri)
298 self.assertEqual(uri.updatedExtension(".txt"), uri)
299 self.assertEqual(id(uri.updatedExtension(".txt")), id(uri))
301 fits = uri.updatedExtension(".fits.gz")
302 self.assertEqual(fits.basename(), "test.fits.gz")
303 self.assertEqual(fits.updatedExtension(".jpeg").basename(), "test.jpeg")
305 extensionless = self.root_uri.join("no_ext")
306 self.assertEqual(extensionless.getExtension(), "")
307 extension = extensionless.updatedExtension(".fits")
308 self.assertEqual(extension.getExtension(), ".fits")
310 uri = ResourcePath("test.txt", forceAbsolute=False)
311 self.assertEqual(uri.getExtension(), ".txt")
312 uri = ResourcePath(self._make_uri("dir.1/dir.2/test.txt"), forceDirectory=False)
313 self.assertEqual(uri.getExtension(), ".txt")
314 uri = ResourcePath(self._make_uri("dir.1/dir.2/"), forceDirectory=True)
315 self.assertEqual(uri.getExtension(), ".2")
316 uri = ResourcePath(self._make_uri("dir.1/dir/"), forceDirectory=True)
317 self.assertEqual(uri.getExtension(), "")
319 def test_relative(self) -> None:
320 """Check that we can get subpaths back from two URIs."""
321 parent = ResourcePath(self._make_uri(self.path1), forceDirectory=True)
322 self.assertTrue(parent.isdir())
323 child = parent.join("dir1/file.txt")
325 self.assertEqual(child.relative_to(parent), "dir1/file.txt")
327 not_child = ResourcePath("/a/b/dir1/file.txt")
328 self.assertIsNone(not_child.relative_to(parent))
329 self.assertFalse(not_child.isdir())
331 not_directory = parent.join("dir1/file2.txt")
332 self.assertIsNone(child.relative_to(not_directory))
334 # Relative URIs
335 parent = ResourcePath("a/b/", forceAbsolute=False)
336 child = ResourcePath("a/b/c/d.txt", forceAbsolute=False)
337 self.assertFalse(child.scheme)
338 self.assertEqual(child.relative_to(parent), "c/d.txt")
340 # forceAbsolute=True should work even on an existing ResourcePath
341 self.assertTrue(pathlib.Path(ResourcePath(child, forceAbsolute=True).ospath).is_absolute())
343 # Absolute URI and schemeless URI
344 parent = self.root_uri.join("/a/b/c/")
345 child = ResourcePath("e/f/g.txt", forceAbsolute=False)
347 # If the child is relative and the parent is absolute we assume
348 # that the child is a child of the parent unless it uses ".."
349 self.assertEqual(child.relative_to(parent), "e/f/g.txt", f"{child}.relative_to({parent})")
351 child = ResourcePath("../e/f/g.txt", forceAbsolute=False)
352 self.assertIsNone(child.relative_to(parent))
354 child = ResourcePath("../c/e/f/g.txt", forceAbsolute=False)
355 self.assertEqual(child.relative_to(parent), "e/f/g.txt")
357 # Test with different netloc
358 child = ResourcePath(self._make_uri("a/b/c.txt", netloc="my.host"))
359 parent = ResourcePath(self._make_uri("a", netloc="other"), forceDirectory=True)
360 self.assertIsNone(child.relative_to(parent), f"{child}.relative_to({parent})")
362 # This is an absolute path so will *always* return a file URI and
363 # ignore the root parameter.
364 parent = ResourcePath("/a/b/c", root=self.root_uri, forceDirectory=True)
365 self.assertEqual(parent.geturl(), "file:///a/b/c/")
367 parent = ResourcePath(self._make_uri("/a/b/c"), forceDirectory=True)
368 child = ResourcePath("d/e.txt", root=parent)
369 self.assertEqual(child.relative_to(parent), "d/e.txt", f"{child}.relative_to({parent})")
371 parent = ResourcePath("c/", root=ResourcePath(self._make_uri("/a/b/")))
372 self.assertEqual(child.relative_to(parent), "d/e.txt", f"{child}.relative_to({parent})")
374 # Absolute schemeless child with relative parent will always fail.
375 child = ResourcePath("d/e.txt", root="/a/b/c")
376 parent = ResourcePath("d/e.txt", forceAbsolute=False)
377 self.assertIsNone(child.relative_to(parent), f"{child}.relative_to({parent})")
379 # Allow .. in response.
380 child = ResourcePath(self._make_uri("a/b/c/d.txt"), forceAbsolute=False)
381 parent = ResourcePath(self._make_uri("a/b/d/e/"), forceAbsolute=False)
382 self.assertIsNone(child.relative_to(parent), f"{child}.relative_to({parent})")
384 if sys.version_info >= (3, 12, 0):
385 # Fails on python 3.11.
386 self.assertEqual(
387 child.relative_to(parent, walk_up=True),
388 "../../c/d.txt",
389 f"{child}.relative_to({parent}, walk_up=True)",
390 )
392 def test_parents(self) -> None:
393 """Test of splitting and parent walking."""
394 parent = ResourcePath(self._make_uri("somedir"), forceDirectory=True)
395 child_file = parent.join("subdir/file.txt")
396 self.assertFalse(child_file.isdir())
397 child_subdir, file = child_file.split()
398 self.assertEqual(file, "file.txt")
399 self.assertTrue(child_subdir.isdir())
400 self.assertEqual(child_file.dirname(), child_subdir)
401 self.assertEqual(child_file.basename(), file)
402 self.assertEqual(child_file.parent(), child_subdir)
403 derived_parent = child_subdir.parent()
404 self.assertEqual(derived_parent, parent)
405 self.assertTrue(derived_parent.isdir())
406 self.assertEqual(child_file.parent().parent(), parent)
407 self.assertEqual(child_subdir.dirname(), child_subdir)
409 # Make sure that the parent doesn't retain any fragment from the
410 # child.
411 child_fragment = child_subdir.join("a.txt#fragment")
412 self.assertEqual(child_fragment.fragment, "fragment")
413 fragment_parent = child_fragment.parent()
414 self.assertEqual(fragment_parent.fragment, "")
415 self.assertTrue(str(fragment_parent).endswith("/"))
417 def test_escapes(self) -> None:
418 """Special characters in file paths."""
419 src = self.root_uri.join("bbb/???/test.txt")
420 self.assertNotIn("???", src.path)
421 self.assertIn("???", src.unquoted_path)
423 file = src.updatedFile("tests??.txt")
424 self.assertNotIn("??.txt", file.path)
426 src = src.updatedFile("tests??.txt")
427 self.assertIn("??.txt", src.unquoted_path)
429 # File URI and schemeless URI
430 parent = ResourcePath(self._make_uri(urllib.parse.quote("/a/b/c/de/??/")))
431 child = ResourcePath("e/f/g.txt", forceAbsolute=False)
432 self.assertEqual(child.relative_to(parent), "e/f/g.txt")
434 child = ResourcePath("e/f??#/g.txt", forceAbsolute=False)
435 self.assertEqual(child.relative_to(parent), "e/f??#/g.txt")
437 child = ResourcePath(self._make_uri(urllib.parse.quote("/a/b/c/de/??/e/f??#/g.txt")))
438 self.assertEqual(child.relative_to(parent), "e/f??#/g.txt")
440 self.assertEqual(child.relativeToPathRoot, "a/b/c/de/??/e/f??#/g.txt")
442 # dir.join() morphs into a file scheme
443 dir = ResourcePath(self._make_uri(urllib.parse.quote("bbb/???/")))
444 new = dir.join("test_j.txt")
445 self.assertIn("???", new.unquoted_path, f"Checking {new}")
447 new2name = "###/test??.txt"
448 new2 = dir.join(new2name)
449 self.assertIn("???", new2.unquoted_path)
450 self.assertTrue(new2.unquoted_path.endswith(new2name))
452 fdir = dir.abspath()
453 self.assertNotIn("???", fdir.path)
454 self.assertIn("???", fdir.unquoted_path)
455 self.assertEqual(fdir.scheme, self.scheme)
457 fnew2 = fdir.join(new2name)
458 self.assertTrue(fnew2.unquoted_path.endswith(new2name))
459 self.assertNotIn("###", fnew2.path)
461 # Test that children relative to schemeless and file schemes
462 # still return the same unquoted name
463 self.assertEqual(fnew2.relative_to(fdir), new2name, f"{fnew2}.relative_to({fdir})")
464 self.assertEqual(fnew2.relative_to(dir), new2name, f"{fnew2}.relative_to({dir})")
465 self.assertEqual(new2.relative_to(fdir), new2name, f"{new2}.relative_to({fdir})")
466 self.assertEqual(new2.relative_to(dir), new2name, f"{new2}.relative_to({dir})")
468 # Check for double quoting
469 plus_path = "/a/b/c+d/"
470 with self.assertLogs(level="WARNING"):
471 uri = ResourcePath(urllib.parse.quote(plus_path), forceDirectory=True)
472 self.assertEqual(uri.ospath, plus_path)
474 # Check that # is not escaped for schemeless URIs
475 hash_path = "/a/b#/c&d#xyz"
476 hpos = hash_path.rfind("#")
477 uri = ResourcePath(hash_path)
478 self.assertEqual(uri.ospath, hash_path[:hpos])
479 self.assertEqual(uri.fragment, hash_path[hpos + 1 :])
480 self.assertEqual(uri.unquoted_fragment, uri.fragment)
482 # Fragments can be quoted, although this is not enforced anywhere.
483 with_frag = ResourcePath(self._make_uri("a/b.txt#" + urllib.parse.quote("zip-path=ingést")))
484 self.assertEqual(with_frag.fragment, "zip-path%3Ding%C3%A9st")
485 self.assertEqual(with_frag.unquoted_fragment, "zip-path=ingést")
487 def test_hash(self) -> None:
488 """Test that we can store URIs in sets and as keys."""
489 uri1 = self.root_uri
490 uri2 = uri1.join("test/")
491 s = {uri1, uri2}
492 self.assertIn(uri1, s)
494 d = {uri1: "1", uri2: "2"}
495 self.assertEqual(d[uri2], "2")
497 def test_root_uri(self) -> None:
498 """Test ResourcePath.root_uri()."""
499 uri = ResourcePath(self._make_uri("a/b/c.txt"))
500 self.assertEqual(uri.root_uri().geturl(), self.root)
502 def test_join(self) -> None:
503 """Test .join method."""
504 root_str = self.root
505 root = self.root_uri
507 self.assertEqual(root.join("b/test.txt").geturl(), f"{root_str}b/test.txt")
508 add_dir = root.join("b/c/d/")
509 self.assertTrue(add_dir.isdir())
510 self.assertEqual(add_dir.geturl(), f"{root_str}b/c/d/")
512 up_relative = root.join("../b/c.txt")
513 self.assertFalse(up_relative.isdir())
514 self.assertEqual(up_relative.geturl(), f"{root_str}b/c.txt")
516 # Check that fragment is passed through join (simple unquoted case).
517 fnew3 = root.join("a/b.txt#fragment")
518 self.assertEqual(fnew3.fragment, "fragment")
519 self.assertEqual(fnew3.basename(), "b.txt", msg=f"Got: {fnew3._uri}")
521 # Check that fragment on the directory is dropped on join.
522 frag_dir = add_dir.join("subdir/#dir_fragment")
523 self.assertEqual(frag_dir.fragment, "dir_fragment")
524 fnew4 = frag_dir.join("a.txt")
525 self.assertEqual(fnew4.fragment, "")
526 self.assertTrue(str(fnew4).endswith("/a.txt"))
528 # Join a resource path.
529 subpath = ResourcePath("a/b.txt#fragment2", forceAbsolute=False, forceDirectory=False)
530 fnew3 = root.join(subpath)
531 self.assertEqual(fnew3.fragment, "fragment2")
532 self.assertEqual(fnew3.basename(), "b.txt", msg=f"Got: {fnew3._uri}")
534 # Quoted string with fragment.
535 quote_example = "hsc/payload/b&c.t@x#t"
536 needs_quote = root.join(quote_example)
537 self.assertEqual(needs_quote.unquoted_path, "/" + quote_example[:-2])
538 self.assertEqual(needs_quote.fragment, "t")
540 other = ResourcePath(f"{self.root}test.txt")
541 self.assertEqual(root.join(other), other)
542 self.assertEqual(other.join("b/new.txt").geturl(), f"{self.root}test.txt/b/new.txt")
544 other = ResourcePath(f"{self.root}text.txt", forceDirectory=False)
545 with self.assertRaises(ValueError):
546 other.join("b/new.text")
548 joined = ResourcePath(f"{self.root}hsc/payload/").join(
549 ResourcePath("test.qgraph", forceAbsolute=False)
550 )
551 self.assertEqual(joined, ResourcePath(f"{self.root}hsc/payload/test.qgraph"))
553 qgraph = ResourcePath("test.qgraph") # Absolute URI
554 joined = ResourcePath(f"{self.root}hsc/payload/").join(qgraph)
555 self.assertEqual(joined, qgraph)
557 with self.assertRaises(ValueError):
558 root.join("dir/", forceDirectory=False)
560 temp = root.join("dir2/", isTemporary=True)
561 with self.assertRaises(RuntimeError):
562 temp.join("test.txt", isTemporary=False)
564 rel = ResourcePath("new.txt", forceAbsolute=False, forceDirectory=False)
565 with self.assertRaises(RuntimeError):
566 root.join(rel, forceDirectory=True)
568 def test_quoting(self) -> None:
569 """Check that quoting works."""
570 parent = ResourcePath(self._make_uri("rootdir"), forceDirectory=True)
571 subpath = "rootdir/dir1+/file?.txt"
572 child = ResourcePath(self._make_uri(urllib.parse.quote(subpath)))
574 self.assertEqual(child.relative_to(parent), "dir1+/file?.txt")
575 self.assertEqual(child.basename(), "file?.txt")
576 self.assertEqual(child.relativeToPathRoot, subpath)
577 self.assertIn("%", child.path)
578 self.assertEqual(child.unquoted_path, "/" + subpath)
580 def test_ordering(self) -> None:
581 """Check that greater/less comparison operators work."""
582 a = self._make_uri("a.txt")
583 b = self._make_uri("b/")
584 self.assertLess(a, b)
585 self.assertFalse(a < a)
586 self.assertLessEqual(a, b)
587 self.assertLessEqual(a, a)
588 self.assertGreater(b, a)
589 self.assertFalse(b > b)
590 self.assertGreaterEqual(b, a)
591 self.assertGreaterEqual(b, b)
594class GenericReadWriteTestCase(_GenericTestCase):
595 """Test schemes that can read and write using concrete resources."""
597 transfer_modes: tuple[str, ...] = ("copy", "move")
598 testdir: str | None = None
599 # Number of files to use for mremove() testing to ensure difference code
600 # paths are hit. Do not want to generically use many files for schemes
601 # where it makes no difference.
602 n_mremove_files: int = 15
604 def setUp(self) -> None:
605 if self.scheme is None:
606 raise unittest.SkipTest("No scheme defined")
607 self.root = self._make_uri("")
608 self.root_uri = ResourcePath(self.root, forceDirectory=True, forceAbsolute=False)
610 if self.scheme == "file":
611 # Use a local tempdir because on macOS the temp dirs use symlinks
612 # so relsymlink gets quite confused.
613 self.tmpdir = ResourcePath(makeTestTempDir(self.testdir), forceDirectory=True)
614 else:
615 # Create random tmp directory relative to the test root.
616 self.tmpdir = self.root_uri.join(
617 "TESTING-" + "".join(random.choices(string.ascii_lowercase + string.digits, k=8)),
618 forceDirectory=True,
619 )
620 self.tmpdir.mkdir()
622 def tearDown(self) -> None:
623 if self.tmpdir and self.tmpdir.isLocal:
624 removeTestTempDir(self.tmpdir.ospath)
626 def test_file(self) -> None:
627 uri = self.tmpdir.join("test.txt")
628 self.assertFalse(uri.exists(), f"{uri} should not exist")
629 self.assertTrue(uri.path.endswith("test.txt"))
631 content = "abcdefghijklmnopqrstuv\n"
632 uri.write(content.encode())
633 self.assertTrue(uri.exists(), f"{uri} should now exist")
634 self.assertEqual(uri.read().decode(), content)
635 self.assertEqual(uri.size(), len(content.encode()))
637 with self.assertRaises(FileExistsError):
638 uri.write(b"", overwrite=False)
640 # Not all backends can tell if a remove fails so we can not
641 # test that a remove of a non-existent entry is guaranteed to raise.
642 uri.remove()
643 self.assertFalse(uri.exists())
645 # Ideally the test would remove the file again and raise a
646 # FileNotFoundError. This is not reliable for remote resources
647 # and doing an explicit check before trying to remove the resource
648 # just to raise an exception is deemed an unacceptable overhead.
650 with self.assertRaises(FileNotFoundError):
651 uri.read()
653 with self.assertRaises(FileNotFoundError):
654 self.tmpdir.join("file/not/there.txt").size()
656 # Check that creating a URI from a URI returns the same thing
657 uri2 = ResourcePath(uri)
658 self.assertEqual(uri, uri2)
659 self.assertEqual(id(uri), id(uri2))
661 def test_mkdir(self) -> None:
662 newdir = self.tmpdir.join("newdir/seconddir", forceDirectory=True)
663 newdir.mkdir()
664 self.assertTrue(newdir.exists())
665 self.assertEqual(newdir.size(), 0)
667 newfile = newdir.join("temp.txt")
668 newfile.write(b"Data")
669 self.assertTrue(newfile.exists())
671 file = self.tmpdir.join("file.txt")
672 # Some schemes will realize that the URI is not a file and so
673 # will raise NotADirectoryError. The file scheme is more permissive
674 # and lets you write anything but will raise NotADirectoryError
675 # if a non-directory is already there. We therefore write something
676 # to the file to ensure that we trigger a portable exception.
677 file.write(b"")
678 with self.assertRaises(NotADirectoryError):
679 file.mkdir()
681 # The root should exist.
682 self.root_uri.mkdir()
683 self.assertTrue(self.root_uri.exists())
685 def test_transfer(self) -> None:
686 src = self.tmpdir.join("test.txt")
687 content = "Content is some content\nwith something to say\n\n"
688 src.write(content.encode())
690 can_move = "move" in self.transfer_modes
691 for mode in self.transfer_modes:
692 if mode == "move":
693 continue
695 dest = self.tmpdir.join(f"dest_{mode}.txt")
696 # Ensure that we get some debugging output.
697 with self.assertLogs("lsst.resources", level=logging.DEBUG) as cm:
698 dest.transfer_from(src, transfer=mode)
699 self.assertIn("Transferring ", "\n".join(cm.output))
700 self.assertTrue(dest.exists(), f"Check that {dest} exists (transfer={mode})")
702 new_content = dest.read().decode()
703 self.assertEqual(new_content, content)
705 if mode in ("symlink", "relsymlink"):
706 self.assertTrue(os.path.islink(dest.ospath), f"Check that {dest} is symlink")
708 # If the source and destination are hardlinks of each other
709 # the transfer should work even if overwrite=False.
710 if mode in ("link", "hardlink"):
711 dest.transfer_from(src, transfer=mode)
712 else:
713 with self.assertRaises(
714 FileExistsError, msg=f"Overwrite of {dest} should not be allowed ({mode})"
715 ):
716 dest.transfer_from(src, transfer=mode)
718 # Transfer again and overwrite.
719 dest.transfer_from(src, transfer=mode, overwrite=True)
721 dest.remove()
723 b = src.read()
724 self.assertEqual(b.decode(), new_content)
726 nbytes = 10
727 subset = src.read(size=nbytes)
728 self.assertEqual(len(subset), nbytes)
729 self.assertEqual(subset.decode(), content[:nbytes])
731 # Transferring to self should be okay.
732 src.transfer_from(src, "auto")
734 with self.assertRaises(ValueError):
735 src.transfer_from(src, transfer="unknown")
737 # A move transfer is special.
738 if can_move:
739 dest.transfer_from(src, transfer="move")
740 self.assertFalse(src.exists())
741 self.assertTrue(dest.exists())
742 else:
743 src.remove()
745 dest.remove()
746 with self.assertRaises(FileNotFoundError):
747 dest.transfer_from(src, "auto")
749 def test_mtransfer(self) -> None:
750 n_files = 10
751 sources = [self.tmpdir.join(f"test{n}.txt") for n in range(n_files)]
752 destinations = [self.tmpdir.join(f"dest_test{n}.txt") for n in range(n_files)]
754 for i, src in enumerate(sources):
755 content = f"{i}\nContent is some content\nwith something to say\n\n"
756 src.write(content.encode())
758 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True))
759 self.assertTrue(all(res.success for res in results.values()))
760 self.assertTrue(all(dest.exists() for dest in results))
762 for i, dest in enumerate(destinations):
763 new_content = dest.read().decode()
764 self.assertTrue(new_content.startswith(f"{i}\n"))
766 # Overwrite should work.
767 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True), overwrite=True)
769 # Overwrite failure.
770 results = ResourcePath.mtransfer(
771 "copy", zip(sources, destinations, strict=True), overwrite=False, do_raise=False
772 )
773 self.assertFalse(all(res.success for res in results.values()))
775 with self.assertRaises(ExceptionGroup):
776 results = ResourcePath.mtransfer(
777 "copy", zip(sources, destinations, strict=True), overwrite=False, do_raise=True
778 )
780 def test_local_transfer(self) -> None:
781 """Test we can transfer to and from local file."""
782 remote_src = self.tmpdir.join("src.json")
783 remote_src.write(b"42")
784 remote_dest = self.tmpdir.join("dest.json")
786 with ResourcePath.temporary_uri(suffix=".json") as tmp:
787 self.assertTrue(tmp.isLocal)
788 tmp.transfer_from(remote_src, transfer="auto")
789 self.assertEqual(tmp.read(), remote_src.read())
791 remote_dest.transfer_from(tmp, transfer="auto")
792 self.assertEqual(remote_dest.read(), tmp.read())
794 # Temporary (possibly remote) resource.
795 # Transfers between temporary resources.
796 with (
797 ResourcePath.temporary_uri(prefix=self.tmpdir.join("tmp"), suffix=".json") as remote_tmp,
798 ResourcePath.temporary_uri(suffix=".json") as local_tmp,
799 ):
800 remote_tmp.write(b"42")
801 if not remote_tmp.isLocal:
802 for transfer in ("link", "symlink", "hardlink", "relsymlink"):
803 with self.assertRaises(RuntimeError):
804 # Trying to symlink a remote resource is not going
805 # to work. A hardlink could work but would rely
806 # on the local temp space being on the same
807 # filesystem as the target.
808 local_tmp.transfer_from(remote_tmp, transfer)
809 local_tmp.transfer_from(remote_tmp, "move")
810 self.assertFalse(remote_tmp.exists())
811 remote_tmp.transfer_from(local_tmp, "auto", overwrite=True)
812 self.assertEqual(local_tmp.read(), remote_tmp.read())
814 # Transfer of missing remote.
815 remote_tmp.remove()
816 with self.assertRaises(FileNotFoundError):
817 local_tmp.transfer_from(remote_tmp, "auto", overwrite=True)
819 def test_local(self) -> None:
820 """Check that remote resources can be made local."""
821 src = self.tmpdir.join("test.txt")
822 original_content = "Content is some content\nwith something to say\n\n"
823 src.write(original_content.encode())
825 # Run this twice to ensure use of cache in code coverage
826 # if applicable.
827 for _ in (1, 2):
828 with src.as_local() as local_uri:
829 self.assertTrue(local_uri.isLocal)
830 content = local_uri.read().decode()
831 self.assertEqual(content, original_content)
833 if src.isLocal:
834 self.assertEqual(src, local_uri)
836 with self.assertRaises(IsADirectoryError):
837 with self.root_uri.as_local() as local_uri:
838 pass
840 if not src.isLocal:
841 # as_local tmpdir can not be a remote resource.
842 with self.assertRaises(ValueError):
843 with src.as_local(tmpdir=self.root_uri) as local_uri:
844 pass
846 # tmpdir is ignored for local file.
847 with tempfile.TemporaryDirectory() as tmpdir:
848 temp_dir = ResourcePath(tmpdir, forceDirectory=True)
849 with src.as_local(tmpdir=temp_dir) as local_uri:
850 self.assertEqual(local_uri.dirname(), temp_dir)
851 self.assertTrue(local_uri.exists())
853 def test_local_mtransfer(self) -> None:
854 """Check that bulk transfer to/from local works."""
855 # Create remote resources
856 n_files = 10
857 sources = [self.tmpdir.join(f"test{n}.txt") for n in range(n_files)]
859 for i, src in enumerate(sources):
860 content = f"{i}\nContent is some content\nwith something to say\n\n"
861 src.write(content.encode())
863 # Potentially remote to local.
864 with tempfile.TemporaryDirectory() as tmpdir:
865 temp_dir = ResourcePath(tmpdir, forceDirectory=True)
866 destinations = [temp_dir.join(f"dest_test{n}.txt") for n in range(n_files)]
868 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True))
869 self.assertTrue(all(res.success for res in results.values()))
870 self.assertTrue(all(dest.exists() for dest in results))
872 # Overwrite should work.
873 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True), overwrite=True)
875 # Now reverse so local to potentially remote.
876 for src in sources:
877 src.remove()
878 results = ResourcePath.mtransfer("copy", zip(destinations, sources, strict=True), overwrite=False)
879 self.assertTrue(all(res.success for res in results.values()))
880 self.assertTrue(all(dest.exists() for dest in results))
882 def test_walk(self) -> None:
883 """Walk a directory hierarchy."""
884 root = self.tmpdir.join("walk/")
886 # Look for a file that is not there
887 file = root.join("config/basic/butler.yaml")
888 found_list = list(ResourcePath.findFileResources([file]))
889 self.assertEqual(found_list[0], file)
891 # First create the files (content is irrelevant).
892 expected_files = {
893 "dir1/a.yaml",
894 "dir1/b.yaml",
895 "dir1/c.json",
896 "dir2/d.json",
897 "dir2/e.yaml",
898 }
899 expected_uris = {root.join(f) for f in expected_files}
900 for uri in expected_uris:
901 uri.write(b"")
902 self.assertTrue(uri.exists())
904 # Look for the files.
905 found = set(ResourcePath.findFileResources([root]))
906 self.assertEqual(found, expected_uris)
908 # Now solely the YAML files.
909 expected_yaml = {u for u in expected_uris if u.getExtension() == ".yaml"}
910 found = set(ResourcePath.findFileResources([root], file_filter=r".*\.yaml$"))
911 self.assertEqual(found, expected_yaml)
913 # Now two explicit directories and a file
914 expected = set(expected_yaml)
915 expected.add(file)
917 found = set(
918 ResourcePath.findFileResources(
919 [file, root.join("dir1/"), root.join("dir2/")],
920 file_filter=r".*\.yaml$",
921 )
922 )
923 self.assertEqual(found, expected)
925 # Group by directory -- find everything and compare it with what
926 # we expected to be there in total.
927 found_yaml = set()
928 counter = 0
929 for uris in ResourcePath.findFileResources([file, root], file_filter=r".*\.yaml$", grouped=True):
930 assert not isinstance(uris, ResourcePath) # for mypy.
931 found_uris = set(uris)
932 if found_uris:
933 counter += 1
935 found_yaml.update(found_uris)
937 expected_yaml_2 = expected_yaml
938 expected_yaml_2.add(file)
939 self.assertEqual(found_yaml, expected_yaml)
940 self.assertEqual(counter, 3)
942 # Grouping but check that single files are returned in a single group
943 # at the end
944 file2 = root.join("config/templates/templates-bad.yaml")
945 found_grouped = [
946 list(group)
947 for group in ResourcePath.findFileResources([file, file2, root.join("dir2/")], grouped=True)
948 if not isinstance(group, ResourcePath) # For mypy.
949 ]
950 self.assertEqual(len(found_grouped), 2, f"Found: {list(found_grouped)}")
951 self.assertEqual(list(found_grouped[1]), [file, file2])
953 with self.assertRaises(ValueError):
954 # The list forces the generator to run.
955 list(file.walk())
957 # A directory that does not exist returns nothing.
958 self.assertEqual(list(root.join("dir3/").walk()), [])
960 def test_large_walk(self) -> None:
961 # In some systems pagination is used so ensure that we can handle
962 # large numbers of files. For example S3 limits us to 1000 responses
963 # per listing call.
964 created = set()
965 counter = 1
966 n_dir1 = 1100
967 root = self.tmpdir.join("large_walk", forceDirectory=True)
968 while counter <= n_dir1:
969 new = ResourcePath(root.join(f"file{counter:04d}.txt"))
970 new.write(f"{counter}".encode())
971 created.add(new)
972 counter += 1
973 counter = 1
974 # Put some in a subdirectory to make sure we are looking in a
975 # hierarchy.
976 n_dir2 = 100
977 subdir = root.join("subdir", forceDirectory=True)
978 while counter <= n_dir2:
979 new = ResourcePath(subdir.join(f"file{counter:04d}.txt"))
980 new.write(f"{counter}".encode())
981 created.add(new)
982 counter += 1
984 found = set(ResourcePath.findFileResources([root]))
985 self.assertEqual(len(found), n_dir1 + n_dir2)
986 self.assertEqual(found, created)
988 # Again with grouping.
989 # (mypy gets upset not knowing which of the two options is being
990 # returned so add useless instance check).
991 found_list = [
992 list(group)
993 for group in ResourcePath.findFileResources([root], grouped=True)
994 if not isinstance(group, ResourcePath) # For mypy.
995 ]
996 self.assertEqual(len(found_list), 2)
997 self.assertEqual(len(found_list[0]), n_dir1)
998 self.assertEqual(len(found_list[1]), n_dir2)
1000 def test_temporary(self) -> None:
1001 prefix = self.tmpdir.join("tmp", forceDirectory=True)
1002 with ResourcePath.temporary_uri(prefix=prefix, suffix=".json") as tmp:
1003 self.assertEqual(tmp.getExtension(), ".json", f"uri: {tmp}")
1004 self.assertTrue(tmp.isabs(), f"uri: {tmp}")
1005 self.assertFalse(tmp.exists(), f"uri: {tmp}")
1006 tmp.write(b"abcd")
1007 self.assertTrue(tmp.exists(), f"uri: {tmp}")
1008 self.assertTrue(tmp.isTemporary)
1009 self.assertFalse(tmp.exists(), f"uri: {tmp}")
1011 tmpdir = ResourcePath(self.tmpdir, forceDirectory=True)
1012 with ResourcePath.temporary_uri(prefix=tmpdir) as tmp:
1013 # Use a specified tmpdir and check it is okay for the file
1014 # to not be created.
1015 self.assertFalse(tmp.getExtension())
1016 self.assertFalse(tmp.exists(), f"uri: {tmp}")
1017 self.assertEqual(tmp.scheme, self.scheme)
1018 self.assertTrue(tmp.isTemporary)
1019 self.assertTrue(tmpdir.exists(), f"uri: {tmpdir} still exists")
1021 # Fake a directory suffix.
1022 with self.assertRaises(NotImplementedError):
1023 with ResourcePath.temporary_uri(prefix=self.root_uri, suffix="xxx/") as tmp:
1024 pass
1026 @unittest.skipIf(fsspec is None, "fsspec is not available.")
1027 def test_fsspec(self) -> None:
1028 """Simple read of a file."""
1029 uri = self.tmpdir.join("test.txt")
1030 self.assertFalse(uri.exists(), f"{uri} should not exist")
1031 self.assertTrue(uri.path.endswith("test.txt"))
1033 content = "abcdefghijklmnopqrstuv\n"
1034 uri.write(content.encode())
1036 try:
1037 fs, path = uri.to_fsspec()
1038 except NotImplementedError as e:
1039 raise unittest.SkipTest(str(e)) from e
1040 except ImportError as e:
1041 # HttpResourcePath.to_fsspec() raises if support
1042 # of fsspec for webDAV back ends is disabled.
1043 raise unittest.SkipTest(str(e)) from e
1044 with fs.open(path, "r") as fd:
1045 as_read = fd.read()
1046 self.assertEqual(as_read, content)
1048 def test_open(self) -> None:
1049 tmpdir = ResourcePath(self.tmpdir, forceDirectory=True)
1050 with ResourcePath.temporary_uri(prefix=tmpdir, suffix=".txt") as tmp:
1051 _check_open(self, tmp, mode_suffixes=("", "t"))
1052 _check_open(self, tmp, mode_suffixes=("t",), encoding="utf-16")
1053 _check_open(self, tmp, mode_suffixes=("t",), prefer_file_temporary=True)
1054 _check_open(self, tmp, mode_suffixes=("t",), encoding="utf-16", prefer_file_temporary=True)
1055 with ResourcePath.temporary_uri(prefix=tmpdir, suffix=".dat") as tmp:
1056 _check_open(self, tmp, mode_suffixes=("b",))
1057 _check_open(self, tmp, mode_suffixes=("b",), prefer_file_temporary=True)
1059 with self.assertRaises(IsADirectoryError):
1060 with self.root_uri.open():
1061 pass
1063 def test_mexists(self) -> None:
1064 root = self.tmpdir.join("mexists/")
1066 # A file that is not there.
1067 file = root.join("config/basic/butler.yaml")
1069 # Create some files. Most schemes the code paths do not change for 10
1070 # vs 1000 files but in some schemes it does.
1071 expected_files = [f"dir1/f{n}.yaml" for n in range(self.n_mremove_files)]
1072 expected_uris = [root.join(f) for f in expected_files]
1073 for uri in expected_uris:
1074 uri.write(b"")
1075 self.assertTrue(uri.exists())
1076 expected_uris.append(file)
1078 # Force to run with fewer workers than there are files.
1079 multi = ResourcePath.mexists(expected_uris, num_workers=3)
1081 for uri, is_there in multi.items():
1082 if uri == file:
1083 self.assertFalse(is_there)
1084 else:
1085 self.assertTrue(is_there)
1087 # Clean up. Unfortunately POSIX raises a FileNotFoundError but
1088 # S3 boto does not complain if there is no key.
1089 ResourcePath.mremove(expected_uris, do_raise=False)
1091 # Check they were really removed.
1092 multi = ResourcePath.mexists(expected_uris, num_workers=3)
1093 for uri, is_there in multi.items():
1094 self.assertFalse(is_there)
1096 # Clean up a subset of files that are already gone, but this can
1097 # trigger a different code path.
1098 ResourcePath.mremove(expected_uris[:5], do_raise=False)