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