Coverage for python/lsst/resources/tests.py: 97%
707 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-22 09:18 +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.
11from __future__ import annotations
13__all__ = ["GenericReadWriteTestCase", "GenericTestCase", "make_remote_test_uri"]
15import datetime
16import logging
17import os
18import pathlib
19import random
20import string
21import sys
22import tempfile
23import unittest
24import urllib.parse
25import uuid
26from collections.abc import Iterable
27from typing import TYPE_CHECKING, Any
29try:
30 import fsspec
31except ImportError:
32 fsspec = None
34from lsst.resources import ResourcePath
35from lsst.resources.utils import makeTestTempDir, removeTestTempDir
37TESTDIR = os.path.abspath(os.path.dirname(__file__))
40def _check_open(
41 test_case: _GenericTestCase | unittest.TestCase,
42 uri: ResourcePath,
43 *,
44 mode_suffixes: Iterable[str] = ("", "t", "b"),
45 **kwargs: Any,
46) -> None:
47 """Test an implementation of ButlerURI.open.
49 Parameters
50 ----------
51 test_case : `unittest.TestCase`
52 Test case to use for assertions.
53 uri : `ResourcePath`
54 URI to use for tests. Must point to a writeable location that is not
55 yet occupied by a file. On return, the location may point to a file
56 only if the test fails.
57 mode_suffixes : `~collections.abc.Iterable` of `str`
58 Suffixes to pass as part of the ``mode`` argument to
59 `ResourcePath.open`, indicating whether to open as binary or as text;
60 the only permitted elements are ``""``, ``"t"``, and ``"b"`.
61 **kwargs
62 Additional keyword arguments to forward to all calls to `open`.
63 """
64 text_content = "abcdefghijklmnopqrstuvwxyz🙂"
65 bytes_content = uuid.uuid4().bytes
66 content_by_mode_suffix: dict[str, str | bytes] = {
67 "": text_content,
68 "t": text_content,
69 "b": bytes_content,
70 }
71 empty_content_by_mode_suffix: dict[str, str | bytes] = {
72 "": "",
73 "t": "",
74 "b": b"",
75 }
76 # To appease mypy
77 double_content_by_mode_suffix: dict[str, str | bytes] = {
78 "": text_content + text_content,
79 "t": text_content + text_content,
80 "b": bytes_content + bytes_content,
81 }
82 for mode_suffix in mode_suffixes:
83 content = content_by_mode_suffix[mode_suffix]
84 double_content = double_content_by_mode_suffix[mode_suffix]
85 # Create file with mode='x', which prohibits overwriting.
86 with uri.open("x" + mode_suffix, **kwargs) as write_buffer:
87 write_buffer.write(content)
88 test_case.assertTrue(uri.exists())
89 # Check that opening with 'x' now raises, and does not modify content.
90 with test_case.assertRaises(FileExistsError):
91 with uri.open("x" + mode_suffix, **kwargs) as write_buffer:
92 write_buffer.write("bad")
93 # Read the file we created and check the contents.
94 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
95 test_case.assertEqual(read_buffer.read(), content)
96 # The names will not match if a local temporary is being written.
97 if not kwargs.get("prefer_file_temporary"):
98 test_case.assertIn(uri.basename(), read_buffer.name)
99 # Check that we can read bytes in a loop and get EOF
100 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
101 # Seek off the end of the file and should read empty back.
102 read_buffer.seek(1024)
103 test_case.assertEqual(read_buffer.tell(), 1024)
104 content_read = read_buffer.read() # Read as much as we can.
105 test_case.assertEqual(len(content_read), 0, f"Read: {content_read!r}, expected empty.")
107 # First read more than the content.
108 read_buffer.seek(0)
109 size = len(content) * 3
110 chunk_read = read_buffer.read(size)
111 test_case.assertEqual(chunk_read, content)
113 # Repeated reads should always return empty string.
114 chunk_read = read_buffer.read(size)
115 test_case.assertEqual(len(chunk_read), 0)
116 chunk_read = read_buffer.read(size)
117 test_case.assertEqual(len(chunk_read), 0)
119 # Go back to start of file and read in smaller chunks.
120 read_buffer.seek(0)
121 size = len(content) // 3
123 content_read = empty_content_by_mode_suffix[mode_suffix]
124 n_reads = 0
125 while chunk_read := read_buffer.read(size):
126 content_read += chunk_read
127 n_reads += 1
128 if n_reads > 10: # In case EOF never hits because of bug.
129 raise AssertionError(
130 f"Failed to stop reading from file after {n_reads} loops. "
131 f"Read {len(content_read)} bytes/characters. Expected {len(content)}."
132 )
133 test_case.assertEqual(content_read, content)
135 # Go back to start of file and read the entire thing.
136 read_buffer.seek(0)
137 content_read = read_buffer.read()
138 test_case.assertEqual(content_read, content)
140 # Seek off the end of the file and should read empty back.
141 # We run this check twice since in some cases the handle will
142 # cache knowledge of the file size.
143 read_buffer.seek(1024)
144 test_case.assertEqual(read_buffer.tell(), 1024)
145 content_read = read_buffer.read()
146 test_case.assertEqual(len(content_read), 0, f"Read: {content_read!r}, expected empty.")
148 # Write multiple chunks with flushing to ensure that any handles that
149 # cache without flushing work properly.
150 n = 3
151 with uri.open("w" + mode_suffix, **kwargs) as write_buffer:
152 for _ in range(n):
153 write_buffer.write(content)
154 write_buffer.flush()
155 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
156 test_case.assertEqual(read_buffer.read(), content * n)
158 # Write two copies of the content, overwriting the single copy there.
159 with uri.open("w" + mode_suffix, **kwargs) as write_buffer:
160 write_buffer.write(double_content)
161 # Read again, this time use mode='r+', which reads what is there and
162 # then lets us write more; we'll use that to reset the file to one
163 # copy of the content.
164 with uri.open("r+" + mode_suffix, **kwargs) as rw_buffer:
165 test_case.assertEqual(rw_buffer.read(), double_content)
166 rw_buffer.seek(0)
167 rw_buffer.truncate()
168 rw_buffer.write(content)
169 rw_buffer.seek(0)
170 test_case.assertEqual(rw_buffer.read(), content)
171 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
172 test_case.assertEqual(read_buffer.read(), content)
173 # Append some more content to the file; should now have two copies.
174 with uri.open("a" + mode_suffix, **kwargs) as append_buffer:
175 append_buffer.write(content)
176 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
177 test_case.assertEqual(read_buffer.read(), double_content)
178 # Final mode to check is w+, which does read/write but truncates first.
179 with uri.open("w+" + mode_suffix, **kwargs) as rw_buffer:
180 test_case.assertEqual(rw_buffer.read(), empty_content_by_mode_suffix[mode_suffix])
181 rw_buffer.write(content)
182 rw_buffer.seek(0)
183 test_case.assertEqual(rw_buffer.read(), content)
184 # Check that two seeks with reads to end return correctly.
185 # Seek is only reliable with "b" mode.
186 if mode_suffix == "b":
187 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
188 size = len(content)
189 seek1 = 2 * size // 3
190 read_buffer.seek(seek1)
191 content1 = read_buffer.read()
192 test_case.assertEqual(len(content1), size - seek1)
193 # Seek earlier and then read to end.
194 seek2 = size // 2
195 read_buffer.seek(seek2)
196 content2 = read_buffer.read()
197 test_case.assertEqual(len(content2), size - seek2)
198 # Check that we can seek from end and read and seek and read.
199 # Negative seek only works in binary mode.
200 with uri.open("rb", **kwargs) as read_buffer:
201 read_buffer.seek(-5, 2) # Relative to end
202 content_read = read_buffer.read()
203 test_case.assertEqual(len(content_read), 5)
204 read_buffer.seek(-10, 2) # Relative to end
205 content_read = read_buffer.read()
206 test_case.assertEqual(len(content_read), 10)
207 with uri.open("r" + mode_suffix, **kwargs) as read_buffer:
208 test_case.assertEqual(read_buffer.read(), content)
209 # Remove file to make room for the next loop of tests with this URI.
210 uri.remove()
213if TYPE_CHECKING:
215 class TestCaseMixin(unittest.TestCase):
216 """Base class for mixin test classes that use TestCase methods."""
218 pass
220else:
222 class TestCaseMixin:
223 """Do-nothing definition of mixin base class for regular execution."""
225 pass
228class _GenericTestCase(TestCaseMixin):
229 """Generic base class for test mixin."""
231 scheme: str | None = None
232 netloc: str | None = None
233 base_path: str | None = None
234 """Path prepended to every URI built by `_make_uri`. Must already be
235 percent-encoded."""
236 path1 = "test_dir"
237 path2 = "file.txt"
239 def _make_uri(self, path: str, netloc: str | None = None) -> str:
240 """Build a URI string for the scheme under test.
242 Parameters
243 ----------
244 path : `str`
245 Path component to place below `base_path`. Must already be
246 percent-encoded.
247 netloc : `str` or `None`, optional
248 Network location to use instead of `netloc`.
250 Returns
251 -------
252 uri : `str`
253 The assembled URI, or ``path`` unchanged if no scheme is defined.
255 Notes
256 -----
257 The components are interpolated into the URI verbatim, so both ``path``
258 and `base_path` must arrive percent-encoded for any scheme whose
259 `~lsst.resources.ResourcePath` class sets ``quotePaths``. This method
260 can not do the encoding itself because it does not know which class
261 will be created from the string it returns, and encoding a path that
262 was already encoded would corrupt it.
264 Getting this wrong is not always visible: `ResourcePath.join` quotes
265 only the component being joined and leaves the rest of the URI alone,
266 so an unencoded ``base_path`` survives into every derived URI, while
267 `~lsst.resources.file.FileResourcePath.walk` rebuilds the URI from the
268 OS path and encodes all of it. The two then disagree for any path
269 holding a character such as ``+`` or a space.
270 """
271 if self.scheme is not None: 271 ↛ 281line 271 didn't jump to line 281 because the condition on line 271 was always true
272 if netloc is None:
273 netloc = self.netloc
274 if path.startswith("/"):
275 path = path[1:]
276 if self.base_path is not None:
277 path = f"{self.base_path}/{path}".lstrip("/")
279 return f"{self.scheme}://{netloc}/{path}"
280 else:
281 return path
284class GenericTestCase(_GenericTestCase):
285 """Test cases for generic manipulation of a `ResourcePath`."""
287 def setUp(self) -> None:
288 if self.scheme is None: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true
289 raise unittest.SkipTest("No scheme defined")
290 self.root = self._make_uri("")
291 self.root_uri = ResourcePath(self.root, forceDirectory=True, forceAbsolute=False)
293 def test_creation(self) -> None:
294 self.assertEqual(self.root_uri.scheme, self.scheme)
295 self.assertEqual(self.root_uri.netloc, self.netloc)
296 self.assertFalse(self.root_uri.query)
297 self.assertFalse(self.root_uri.params)
299 with self.assertRaises(ValueError):
300 ResourcePath({}) # type: ignore
302 with self.assertRaises(RuntimeError):
303 ResourcePath(self.root_uri, isTemporary=True)
305 file = self.root_uri.join("file.txt", forceDirectory=False)
306 with self.assertRaises(RuntimeError):
307 ResourcePath(file, forceDirectory=True)
309 file = self.root_uri.join("file.txt")
310 file_as_dir = ResourcePath(file, forceDirectory=True)
311 self.assertTrue(file_as_dir.isdir())
313 dir = self._make_uri("a/b/c/")
314 with self.assertRaises(ValueError):
315 ResourcePath(dir, forceDirectory=False)
317 with self.assertRaises(NotImplementedError):
318 ResourcePath("unknown://netloc")
320 replaced = file.replace(fragment="frag")
321 self.assertEqual(replaced.fragment, "frag")
323 with self.assertRaises(ValueError):
324 file.replace(scheme="new")
326 self.assertNotEqual(replaced, str(replaced))
327 self.assertNotEqual(str(replaced), replaced)
329 def test_extension(self) -> None:
330 uri = ResourcePath(self._make_uri("dir/test.txt"))
331 self.assertEqual(uri.updatedExtension(None), uri)
332 self.assertEqual(uri.updatedExtension(".txt"), uri)
333 self.assertEqual(id(uri.updatedExtension(".txt")), id(uri))
335 fits = uri.updatedExtension(".fits.gz")
336 self.assertEqual(fits.basename(), "test.fits.gz")
337 self.assertEqual(fits.updatedExtension(".jpeg").basename(), "test.jpeg")
339 extensionless = self.root_uri.join("no_ext")
340 self.assertEqual(extensionless.getExtension(), "")
341 extension = extensionless.updatedExtension(".fits")
342 self.assertEqual(extension.getExtension(), ".fits")
344 uri = ResourcePath("test.txt", forceAbsolute=False)
345 self.assertEqual(uri.getExtension(), ".txt")
346 uri = ResourcePath(self._make_uri("dir.1/dir.2/test.txt"), forceDirectory=False)
347 self.assertEqual(uri.getExtension(), ".txt")
348 uri = ResourcePath(self._make_uri("dir.1/dir.2/"), forceDirectory=True)
349 self.assertEqual(uri.getExtension(), ".2")
350 uri = ResourcePath(self._make_uri("dir.1/dir/"), forceDirectory=True)
351 self.assertEqual(uri.getExtension(), "")
353 def test_relative(self) -> None:
354 """Check that we can get subpaths back from two URIs."""
355 parent = ResourcePath(self._make_uri(self.path1), forceDirectory=True)
356 self.assertTrue(parent.isdir())
357 child = parent.join("dir1/file.txt")
359 self.assertEqual(child.relative_to(parent), "dir1/file.txt")
361 not_child = ResourcePath("/a/b/dir1/file.txt")
362 self.assertIsNone(not_child.relative_to(parent))
363 self.assertFalse(not_child.isdir())
365 not_directory = parent.join("dir1/file2.txt")
366 self.assertIsNone(child.relative_to(not_directory))
368 # Relative URIs
369 parent = ResourcePath("a/b/", forceAbsolute=False)
370 child = ResourcePath("a/b/c/d.txt", forceAbsolute=False)
371 self.assertFalse(child.scheme)
372 self.assertEqual(child.relative_to(parent), "c/d.txt")
374 # forceAbsolute=True should work even on an existing ResourcePath
375 self.assertTrue(pathlib.Path(ResourcePath(child, forceAbsolute=True).ospath).is_absolute())
377 # Absolute URI and schemeless URI
378 parent = self.root_uri.join("/a/b/c/")
379 child = ResourcePath("e/f/g.txt", forceAbsolute=False)
381 # If the child is relative and the parent is absolute we assume
382 # that the child is a child of the parent unless it uses ".."
383 self.assertEqual(child.relative_to(parent), "e/f/g.txt", f"{child}.relative_to({parent})")
385 child = ResourcePath("../e/f/g.txt", forceAbsolute=False)
386 self.assertIsNone(child.relative_to(parent))
388 child = ResourcePath("../c/e/f/g.txt", forceAbsolute=False)
389 self.assertEqual(child.relative_to(parent), "e/f/g.txt")
391 # Test with different netloc
392 child = ResourcePath(self._make_uri("a/b/c.txt", netloc="my.host"))
393 parent = ResourcePath(self._make_uri("a", netloc="other"), forceDirectory=True)
394 self.assertIsNone(child.relative_to(parent), f"{child}.relative_to({parent})")
396 # This is an absolute path so will *always* return a file URI and
397 # ignore the root parameter.
398 parent = ResourcePath("/a/b/c", root=self.root_uri, forceDirectory=True)
399 self.assertEqual(parent.geturl(), "file:///a/b/c/")
401 parent = ResourcePath(self._make_uri("/a/b/c"), forceDirectory=True)
402 child = ResourcePath("d/e.txt", root=parent)
403 self.assertEqual(child.relative_to(parent), "d/e.txt", f"{child}.relative_to({parent})")
405 parent = ResourcePath("c/", root=ResourcePath(self._make_uri("/a/b/")))
406 self.assertEqual(child.relative_to(parent), "d/e.txt", f"{child}.relative_to({parent})")
408 # Absolute schemeless child with relative parent will always fail.
409 child = ResourcePath("d/e.txt", root="/a/b/c")
410 parent = ResourcePath("d/e.txt", forceAbsolute=False)
411 self.assertIsNone(child.relative_to(parent), f"{child}.relative_to({parent})")
413 # Allow .. in response.
414 child = ResourcePath(self._make_uri("a/b/c/d.txt"), forceAbsolute=False)
415 parent = ResourcePath(self._make_uri("a/b/d/e/"), forceAbsolute=False)
416 self.assertIsNone(child.relative_to(parent), f"{child}.relative_to({parent})")
418 if sys.version_info >= (3, 12, 0): 418 ↛ exitline 418 didn't return from function 'test_relative' because the condition on line 418 was always true
419 # Fails on python 3.11.
420 self.assertEqual(
421 child.relative_to(parent, walk_up=True),
422 "../../c/d.txt",
423 f"{child}.relative_to({parent}, walk_up=True)",
424 )
426 def test_parents(self) -> None:
427 """Test of splitting and parent walking."""
428 parent = ResourcePath(self._make_uri("somedir"), forceDirectory=True)
429 child_file = parent.join("subdir/file.txt")
430 self.assertFalse(child_file.isdir())
431 child_subdir, file = child_file.split()
432 self.assertEqual(file, "file.txt")
433 self.assertTrue(child_subdir.isdir())
434 self.assertEqual(child_file.dirname(), child_subdir)
435 self.assertEqual(child_file.basename(), file)
436 self.assertEqual(child_file.parent(), child_subdir)
437 derived_parent = child_subdir.parent()
438 self.assertEqual(derived_parent, parent)
439 self.assertTrue(derived_parent.isdir())
440 self.assertEqual(child_file.parent().parent(), parent)
441 self.assertEqual(child_subdir.dirname(), child_subdir)
443 # Make sure that the parent doesn't retain any fragment from the
444 # child.
445 child_fragment = child_subdir.join("a.txt#fragment")
446 self.assertEqual(child_fragment.fragment, "fragment")
447 fragment_parent = child_fragment.parent()
448 self.assertEqual(fragment_parent.fragment, "")
449 self.assertTrue(str(fragment_parent).endswith("/"))
451 def test_escapes(self) -> None:
452 """Special characters in file paths."""
453 src = self.root_uri.join("bbb/???/test.txt")
454 quotes = src.quotePaths
456 if quotes:
457 self.assertNotIn("???", src.path)
458 else:
459 self.assertIn("???", src.path)
460 self.assertIn("???", src.unquoted_path)
462 file = src.updatedFile("tests??.txt")
463 if quotes:
464 self.assertNotIn("??.txt", file.path)
465 else:
466 self.assertIn("??.txt", file.path)
468 src = src.updatedFile("tests??.txt")
469 self.assertIn("??.txt", src.unquoted_path)
471 # File URI and schemeless URI
472 parent = ResourcePath(self._make_uri(urllib.parse.quote("/a/b/c/de/??/")))
473 child = ResourcePath("e/f/g.txt", forceAbsolute=False)
474 self.assertEqual(child.relative_to(parent), "e/f/g.txt")
476 child = ResourcePath("e/f??#/g.txt", forceAbsolute=False)
477 self.assertEqual(child.relative_to(parent), "e/f??#/g.txt")
479 child = ResourcePath(self._make_uri(urllib.parse.quote("/a/b/c/de/??/e/f??#/g.txt")))
480 self.assertEqual(child.relative_to(parent), "e/f??#/g.txt")
482 self.assertEqual(child.relativeToPathRoot, "a/b/c/de/??/e/f??#/g.txt")
484 # dir.join() morphs into a file scheme
485 dir = ResourcePath(self._make_uri(urllib.parse.quote("bbb/???/")))
486 new = dir.join("test_j.txt")
487 self.assertIn("???", new.unquoted_path, f"Checking {new}")
489 new2name = "###/test??.txt"
490 new2 = dir.join(new2name)
491 self.assertIn("???", new2.unquoted_path)
492 self.assertTrue(new2.unquoted_path.endswith(new2name))
494 fdir = dir.abspath()
495 self.assertNotIn("???", fdir.path)
496 self.assertIn("???", fdir.unquoted_path)
497 self.assertEqual(fdir.scheme, self.scheme)
499 fnew2 = fdir.join(new2name)
500 self.assertTrue(fnew2.unquoted_path.endswith(new2name))
501 if quotes:
502 self.assertNotIn("###", fnew2.path)
503 else:
504 self.assertIn("###", fnew2.path)
506 # Test that children relative to schemeless and file schemes
507 # still return the same unquoted name
508 self.assertEqual(fnew2.relative_to(fdir), new2name, f"{fnew2}.relative_to({fdir})")
509 self.assertEqual(fnew2.relative_to(dir), new2name, f"{fnew2}.relative_to({dir})")
510 self.assertEqual(new2.relative_to(fdir), new2name, f"{new2}.relative_to({fdir})")
511 self.assertEqual(new2.relative_to(dir), new2name, f"{new2}.relative_to({dir})")
513 # Check for double quoting
514 plus_path = "/a/b/c+d/"
515 with self.assertLogs(level="WARNING"):
516 uri = ResourcePath(urllib.parse.quote(plus_path), forceDirectory=True)
517 self.assertEqual(uri.ospath, plus_path)
519 # Check that # is not escaped for schemeless URIs
520 hash_path = "/a/b#/c&d#xyz"
521 hpos = hash_path.rfind("#")
522 uri = ResourcePath(hash_path)
523 self.assertEqual(uri.ospath, hash_path[:hpos])
524 self.assertEqual(uri.fragment, hash_path[hpos + 1 :])
525 self.assertEqual(uri.unquoted_fragment, uri.fragment)
527 # Fragments can be quoted, although this is not enforced anywhere.
528 with_frag = ResourcePath(self._make_uri("a/b.txt#" + urllib.parse.quote("zip-path=ingést")))
529 self.assertEqual(with_frag.fragment, "zip-path%3Ding%C3%A9st")
530 self.assertEqual(with_frag.unquoted_fragment, "zip-path=ingést")
532 def test_hash(self) -> None:
533 """Test that we can store URIs in sets and as keys."""
534 uri1 = self.root_uri
535 uri2 = uri1.join("test/")
536 s = {uri1, uri2}
537 self.assertIn(uri1, s)
539 d = {uri1: "1", uri2: "2"}
540 self.assertEqual(d[uri2], "2")
542 def test_root_uri(self) -> None:
543 """Test ResourcePath.root_uri()."""
544 uri = ResourcePath(self._make_uri("a/b/c.txt"))
545 self.assertEqual(uri.root_uri().geturl(), self.root)
547 def test_join(self) -> None:
548 """Test .join method."""
549 root_str = self.root
550 root = self.root_uri
552 self.assertEqual(root.join("b/test.txt").geturl(), f"{root_str}b/test.txt")
553 add_dir = root.join("b/c/d/")
554 self.assertTrue(add_dir.isdir())
555 self.assertEqual(add_dir.geturl(), f"{root_str}b/c/d/")
557 up_relative = root.join("../b/c.txt")
558 self.assertFalse(up_relative.isdir())
559 self.assertEqual(up_relative.geturl(), f"{root_str}b/c.txt")
561 # Check that fragment is passed through join (simple unquoted case).
562 fnew3 = root.join("a/b.txt#fragment")
563 self.assertEqual(fnew3.fragment, "fragment")
564 self.assertEqual(fnew3.basename(), "b.txt", msg=f"Got: {fnew3._uri}")
566 # Check that fragment on the directory is dropped on join.
567 frag_dir = add_dir.join("subdir/#dir_fragment")
568 self.assertEqual(frag_dir.fragment, "dir_fragment")
569 fnew4 = frag_dir.join("a.txt")
570 self.assertEqual(fnew4.fragment, "")
571 self.assertTrue(str(fnew4).endswith("/a.txt"))
573 # Join a resource path.
574 subpath = ResourcePath("a/b.txt#fragment2", forceAbsolute=False, forceDirectory=False)
575 fnew3 = root.join(subpath)
576 self.assertEqual(fnew3.fragment, "fragment2")
577 self.assertEqual(fnew3.basename(), "b.txt", msg=f"Got: {fnew3._uri}")
579 # Quoted string with fragment.
580 quote_example = "hsc/payload/b&c.t@x#t"
581 needs_quote = root.join(quote_example)
582 self.assertEqual(needs_quote.unquoted_path, "/" + quote_example[:-2])
583 self.assertEqual(needs_quote.fragment, "t")
585 other = ResourcePath(f"{self.root}test.txt")
586 self.assertEqual(root.join(other), other)
587 self.assertEqual(other.join("b/new.txt").geturl(), f"{self.root}test.txt/b/new.txt")
589 other = ResourcePath(f"{self.root}text.txt", forceDirectory=False)
590 with self.assertRaises(ValueError):
591 other.join("b/new.text")
593 joined = ResourcePath(f"{self.root}hsc/payload/").join(
594 ResourcePath("test.qgraph", forceAbsolute=False)
595 )
596 self.assertEqual(joined, ResourcePath(f"{self.root}hsc/payload/test.qgraph"))
598 qgraph = ResourcePath("test.qgraph") # Absolute URI
599 joined = ResourcePath(f"{self.root}hsc/payload/").join(qgraph)
600 self.assertEqual(joined, qgraph)
602 with self.assertRaises(ValueError):
603 root.join("dir/", forceDirectory=False)
605 temp = root.join("dir2/", isTemporary=True)
606 with self.assertRaises(RuntimeError):
607 temp.join("test.txt", isTemporary=False)
609 rel = ResourcePath("new.txt", forceAbsolute=False, forceDirectory=False)
610 with self.assertRaises(RuntimeError):
611 root.join(rel, forceDirectory=True)
613 def test_quoting(self) -> None:
614 """Check that quoting works."""
615 parent = ResourcePath(self._make_uri("rootdir"), forceDirectory=True)
616 subpath = "rootdir/dir1+/file?.txt"
617 child = ResourcePath(self._make_uri(urllib.parse.quote(subpath)))
619 self.assertEqual(child.relative_to(parent), "dir1+/file?.txt")
620 self.assertEqual(child.basename(), "file?.txt")
621 self.assertEqual(child.relativeToPathRoot, subpath)
622 self.assertIn("%", child.path)
623 self.assertEqual(child.unquoted_path, "/" + subpath)
625 def test_ordering(self) -> None:
626 """Check that greater/less comparison operators work."""
627 a = self._make_uri("a.txt")
628 b = self._make_uri("b/")
629 self.assertLess(a, b)
630 self.assertFalse(a < a)
631 self.assertLessEqual(a, b)
632 self.assertLessEqual(a, a)
633 self.assertGreater(b, a)
634 self.assertFalse(b > b)
635 self.assertGreaterEqual(b, a)
636 self.assertGreaterEqual(b, b)
639class GenericReadWriteTestCase(_GenericTestCase):
640 """Test schemes that can read and write using concrete resources."""
642 transfer_modes: tuple[str, ...] = ("copy", "move")
643 testdir: str | None = None
644 # Number of files to use for mremove() testing to ensure difference code
645 # paths are hit. Do not want to generically use many files for schemes
646 # where it makes no difference.
647 n_mremove_files: int = 15
649 def setUp(self) -> None:
650 if self.scheme is None: 650 ↛ 651line 650 didn't jump to line 651 because the condition on line 650 was never true
651 raise unittest.SkipTest("No scheme defined")
652 self.root = self._make_uri("")
653 self.root_uri = ResourcePath(self.root, forceDirectory=True, forceAbsolute=False)
655 if self.scheme == "file":
656 # Use a local tempdir because on macOS the temp dirs use symlinks
657 # so relsymlink gets quite confused.
658 self.tmpdir = ResourcePath(makeTestTempDir(self.testdir), forceDirectory=True)
659 else:
660 # Create random tmp directory relative to the test root.
661 self.tmpdir = self.root_uri.join(
662 "TESTING-" + "".join(random.choices(string.ascii_lowercase + string.digits, k=8)),
663 forceDirectory=True,
664 )
665 self.tmpdir.mkdir()
667 def tearDown(self) -> None:
668 if self.tmpdir and self.tmpdir.isLocal:
669 removeTestTempDir(self.tmpdir.ospath)
671 def test_file(self) -> None:
672 uri = self.tmpdir.join("test.txt")
673 self.assertFalse(uri.exists(), f"{uri} should not exist")
674 self.assertTrue(uri.path.endswith("test.txt"))
676 content = "abcdefghijklmnopqrstuv\n"
677 uri.write(content.encode())
678 self.assertTrue(uri.exists(), f"{uri} should now exist")
679 self.assertEqual(uri.read().decode(), content)
680 self.assertEqual(uri.size(), len(content.encode()))
682 with self.assertRaises(FileExistsError):
683 uri.write(b"", overwrite=False)
685 # Not all backends can tell if a remove fails so we can not
686 # test that a remove of a non-existent entry is guaranteed to raise.
687 uri.remove()
688 self.assertFalse(uri.exists())
690 # Ideally the test would remove the file again and raise a
691 # FileNotFoundError. This is not reliable for remote resources
692 # and doing an explicit check before trying to remove the resource
693 # just to raise an exception is deemed an unacceptable overhead.
695 with self.assertRaises(FileNotFoundError):
696 uri.read()
698 with self.assertRaises(FileNotFoundError):
699 self.tmpdir.join("file/not/there.txt").size()
701 # Check that creating a URI from a URI returns the same thing
702 uri2 = ResourcePath(uri)
703 self.assertEqual(uri, uri2)
704 self.assertEqual(id(uri), id(uri2))
706 def test_get_info_generic(self) -> None:
707 """Test generic get_info properties."""
708 now = datetime.datetime.now(tz=datetime.UTC)
709 uri = self.tmpdir.join("test.txt")
711 with self.assertRaises(FileNotFoundError):
712 uri.get_info()
714 content = "abcdefghijklmnopqrstuv\n"
715 uri.write(content.encode())
717 info = uri.get_info()
718 self.assertTrue(info.is_file)
719 self.assertEqual(info.size, len(content))
720 assert info.last_modified is not None
721 self.assertGreaterEqual(info.last_modified.timestamp(), now.timestamp() - 1.0)
722 self.assertIsInstance(info.checksums, dict) # Checksums are backend dependent.
724 for dir_uri in (uri.parent(), uri.root_uri()):
725 # File URIs can return values for modification dates for
726 # directories. S3 URIs can return checksums for directories.
727 dirinfo = dir_uri.get_info()
728 self.assertEqual(dirinfo.uri, str(dir_uri))
729 self.assertFalse(dirinfo.is_file)
730 self.assertEqual(dirinfo.size, 0)
732 newdir = self.tmpdir.join("newdir/", forceDirectory=True)
733 with self.assertRaises(FileNotFoundError):
734 newdir.get_info()
736 def test_mkdir(self) -> None:
737 newdir = self.tmpdir.join("newdir/seconddir", forceDirectory=True)
738 newdir.mkdir()
739 self.assertTrue(newdir.exists())
740 self.assertEqual(newdir.size(), 0)
742 newfile = newdir.join("temp.txt")
743 newfile.write(b"Data")
744 self.assertTrue(newfile.exists())
746 file = self.tmpdir.join("file.txt")
747 # Some schemes will realize that the URI is not a file and so
748 # will raise NotADirectoryError. The file scheme is more permissive
749 # and lets you write anything but will raise NotADirectoryError
750 # if a non-directory is already there. We therefore write something
751 # to the file to ensure that we trigger a portable exception.
752 file.write(b"")
753 with self.assertRaises(NotADirectoryError):
754 file.mkdir()
756 # The root should exist.
757 self.root_uri.mkdir()
758 self.assertTrue(self.root_uri.exists())
760 def test_transfer(self) -> None:
761 src = self.tmpdir.join("test.txt")
762 content = "Content is some content\nwith something to say\n\n"
763 src.write(content.encode())
765 can_move = "move" in self.transfer_modes
766 for mode in self.transfer_modes:
767 if mode == "move":
768 continue
770 dest = self.tmpdir.join(f"dest_{mode}.txt")
771 # Ensure that we get some debugging output.
772 with self.assertLogs("lsst.resources", level=logging.DEBUG) as cm:
773 dest.transfer_from(src, transfer=mode)
774 self.assertIn("Transferring ", "\n".join(cm.output))
775 self.assertTrue(dest.exists(), f"Check that {dest} exists (transfer={mode})")
777 new_content = dest.read().decode()
778 self.assertEqual(new_content, content)
780 if mode in ("symlink", "relsymlink"):
781 self.assertTrue(os.path.islink(dest.ospath), f"Check that {dest} is symlink")
783 # If the source and destination are hardlinks of each other
784 # the transfer should work even if overwrite=False.
785 if mode in ("link", "hardlink"):
786 dest.transfer_from(src, transfer=mode)
787 else:
788 with self.assertRaises(
789 FileExistsError, msg=f"Overwrite of {dest} should not be allowed ({mode})"
790 ):
791 dest.transfer_from(src, transfer=mode)
793 # Transfer again and overwrite.
794 dest.transfer_from(src, transfer=mode, overwrite=True)
796 dest.remove()
798 b = src.read()
799 self.assertEqual(b.decode(), new_content)
801 nbytes = 10
802 subset = src.read(size=nbytes)
803 self.assertEqual(len(subset), nbytes)
804 self.assertEqual(subset.decode(), content[:nbytes])
806 # Transferring to self should be okay.
807 src.transfer_from(src, "auto")
809 with self.assertRaises(ValueError):
810 src.transfer_from(src, transfer="unknown")
812 # A move transfer is special.
813 if can_move: 813 ↛ 818line 813 didn't jump to line 818 because the condition on line 813 was always true
814 dest.transfer_from(src, transfer="move")
815 self.assertFalse(src.exists())
816 self.assertTrue(dest.exists())
817 else:
818 src.remove()
820 dest.remove()
821 with self.assertRaises(FileNotFoundError):
822 dest.transfer_from(src, "auto")
824 def test_mtransfer(self) -> None:
825 n_files = 10
826 sources = [self.tmpdir.join(f"test{n}.txt") for n in range(n_files)]
827 destinations = [self.tmpdir.join(f"dest_test{n}.txt") for n in range(n_files)]
829 for i, src in enumerate(sources):
830 content = f"{i}\nContent is some content\nwith something to say\n\n"
831 src.write(content.encode())
833 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True))
834 self.assertTrue(all(res.success for res in results.values()))
835 self.assertTrue(all(dest.exists() for dest in results))
837 for i, dest in enumerate(destinations):
838 new_content = dest.read().decode()
839 self.assertTrue(new_content.startswith(f"{i}\n"))
841 # Overwrite should work.
842 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True), overwrite=True)
844 # Overwrite failure.
845 results = ResourcePath.mtransfer(
846 "copy", zip(sources, destinations, strict=True), overwrite=False, do_raise=False
847 )
848 self.assertFalse(all(res.success for res in results.values()))
850 with self.assertRaises(ExceptionGroup):
851 results = ResourcePath.mtransfer(
852 "copy", zip(sources, destinations, strict=True), overwrite=False, do_raise=True
853 )
855 def test_local_transfer(self) -> None:
856 """Test we can transfer to and from local file."""
857 remote_src = self.tmpdir.join("src.json")
858 remote_src.write(b"42")
859 remote_dest = self.tmpdir.join("dest.json")
861 with ResourcePath.temporary_uri(suffix=".json") as tmp:
862 self.assertTrue(tmp.isLocal)
863 tmp.transfer_from(remote_src, transfer="auto")
864 self.assertEqual(tmp.read(), remote_src.read())
866 remote_dest.transfer_from(tmp, transfer="auto")
867 self.assertEqual(remote_dest.read(), tmp.read())
869 # Temporary (possibly remote) resource.
870 # Transfers between temporary resources.
871 with (
872 ResourcePath.temporary_uri(prefix=self.tmpdir.join("tmp"), suffix=".json") as remote_tmp,
873 ResourcePath.temporary_uri(suffix=".json") as local_tmp,
874 ):
875 remote_tmp.write(b"42")
876 if not remote_tmp.isLocal:
877 for transfer in ("link", "symlink", "hardlink", "relsymlink"):
878 with self.assertRaises(RuntimeError):
879 # Trying to symlink a remote resource is not going
880 # to work. A hardlink could work but would rely
881 # on the local temp space being on the same
882 # filesystem as the target.
883 local_tmp.transfer_from(remote_tmp, transfer)
884 local_tmp.transfer_from(remote_tmp, "move")
885 self.assertFalse(remote_tmp.exists())
886 remote_tmp.transfer_from(local_tmp, "auto", overwrite=True)
887 self.assertEqual(local_tmp.read(), remote_tmp.read())
889 # Transfer of missing remote.
890 remote_tmp.remove()
891 with self.assertRaises(FileNotFoundError):
892 local_tmp.transfer_from(remote_tmp, "auto", overwrite=True)
894 def test_local(self) -> None:
895 """Check that remote resources can be made local."""
896 src = self.tmpdir.join("test.txt")
897 original_content = "Content is some content\nwith something to say\n\n"
898 src.write(original_content.encode())
900 # Run this twice to ensure use of cache in code coverage
901 # if applicable.
902 for _ in (1, 2):
903 with src.as_local() as local_uri:
904 self.assertTrue(local_uri.isLocal)
905 content = local_uri.read().decode()
906 self.assertEqual(content, original_content)
908 if src.isLocal:
909 self.assertEqual(src, local_uri)
911 with self.assertRaises(IsADirectoryError):
912 with self.root_uri.as_local() as local_uri:
913 pass
915 if not src.isLocal:
916 # as_local tmpdir can not be a remote resource.
917 with self.assertRaises(ValueError):
918 with src.as_local(tmpdir=self.root_uri) as local_uri:
919 pass
921 # tmpdir is ignored for local file.
922 with tempfile.TemporaryDirectory() as tmpdir:
923 temp_dir = ResourcePath(tmpdir, forceDirectory=True)
924 with src.as_local(tmpdir=temp_dir) as local_uri:
925 self.assertEqual(local_uri.dirname(), temp_dir)
926 self.assertTrue(local_uri.exists())
928 def test_local_mtransfer(self) -> None:
929 """Check that bulk transfer to/from local works."""
930 # Create remote resources
931 n_files = 10
932 sources = [self.tmpdir.join(f"test{n}.txt") for n in range(n_files)]
934 for i, src in enumerate(sources):
935 content = f"{i}\nContent is some content\nwith something to say\n\n"
936 src.write(content.encode())
938 # Potentially remote to local.
939 with tempfile.TemporaryDirectory() as tmpdir:
940 temp_dir = ResourcePath(tmpdir, forceDirectory=True)
941 destinations = [temp_dir.join(f"dest_test{n}.txt") for n in range(n_files)]
943 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True))
944 self.assertTrue(all(res.success for res in results.values()))
945 self.assertTrue(all(dest.exists() for dest in results))
947 # Overwrite should work.
948 results = ResourcePath.mtransfer("copy", zip(sources, destinations, strict=True), overwrite=True)
950 # Now reverse so local to potentially remote.
951 for src in sources:
952 src.remove()
953 results = ResourcePath.mtransfer("copy", zip(destinations, sources, strict=True), overwrite=False)
954 self.assertTrue(all(res.success for res in results.values()))
955 self.assertTrue(all(dest.exists() for dest in results))
957 def test_walk(self) -> None:
958 """Walk a directory hierarchy."""
959 root = self.tmpdir.join("walk/")
961 # Look for a file that is not there
962 file = root.join("config/basic/butler.yaml")
963 found_list = list(ResourcePath.findFileResources([file]))
964 self.assertEqual(found_list[0], file)
966 # First create the files (content is irrelevant).
967 expected_files = {
968 "dir1/a.yaml",
969 "dir1/b.yaml",
970 "dir1/c.json",
971 "dir2/d.json",
972 "dir2/e.yaml",
973 }
974 expected_uris = {root.join(f) for f in expected_files}
975 for uri in expected_uris:
976 uri.write(b"")
977 self.assertTrue(uri.exists())
979 # Look for the files.
980 found = set(ResourcePath.findFileResources([root]))
981 self.assertEqual(found, expected_uris)
983 # Now solely the YAML files.
984 expected_yaml = {u for u in expected_uris if u.getExtension() == ".yaml"}
985 found = set(ResourcePath.findFileResources([root], file_filter=r".*\.yaml$"))
986 self.assertEqual(found, expected_yaml)
988 # Now two explicit directories and a file
989 expected = set(expected_yaml)
990 expected.add(file)
992 found = set(
993 ResourcePath.findFileResources(
994 [file, root.join("dir1/"), root.join("dir2/")],
995 file_filter=r".*\.yaml$",
996 )
997 )
998 self.assertEqual(found, expected)
1000 # Group by directory -- find everything and compare it with what
1001 # we expected to be there in total.
1002 found_yaml = set()
1003 counter = 0
1004 for uris in ResourcePath.findFileResources([file, root], file_filter=r".*\.yaml$", grouped=True):
1005 assert not isinstance(uris, ResourcePath) # for mypy.
1006 found_uris = set(uris)
1007 if found_uris: 1007 ↛ 1010line 1007 didn't jump to line 1010 because the condition on line 1007 was always true
1008 counter += 1
1010 found_yaml.update(found_uris)
1012 expected_yaml_2 = expected_yaml
1013 expected_yaml_2.add(file)
1014 self.assertEqual(found_yaml, expected_yaml)
1015 self.assertEqual(counter, 3)
1017 # Grouping but check that single files are returned in a single group
1018 # at the end
1019 file2 = root.join("config/templates/templates-bad.yaml")
1020 found_grouped = [
1021 list(group)
1022 for group in ResourcePath.findFileResources([file, file2, root.join("dir2/")], grouped=True)
1023 if not isinstance(group, ResourcePath) # For mypy.
1024 ]
1025 self.assertEqual(len(found_grouped), 2, f"Found: {list(found_grouped)}")
1026 self.assertEqual(list(found_grouped[1]), [file, file2])
1028 with self.assertRaises(ValueError):
1029 # The list forces the generator to run.
1030 list(file.walk())
1032 # A directory that does not exist returns nothing.
1033 self.assertEqual(list(root.join("dir3/").walk()), [])
1035 def test_large_walk(self) -> None:
1036 # In some systems pagination is used so ensure that we can handle
1037 # large numbers of files. For example S3 limits us to 1000 responses
1038 # per listing call.
1039 created = set()
1040 counter = 1
1041 n_dir1 = 1100
1042 root = self.tmpdir.join("large_walk", forceDirectory=True)
1043 while counter <= n_dir1:
1044 new = ResourcePath(root.join(f"file{counter:04d}.txt"))
1045 new.write(f"{counter}".encode())
1046 created.add(new)
1047 counter += 1
1048 counter = 1
1049 # Put some in a subdirectory to make sure we are looking in a
1050 # hierarchy.
1051 n_dir2 = 100
1052 subdir = root.join("subdir", forceDirectory=True)
1053 while counter <= n_dir2:
1054 new = ResourcePath(subdir.join(f"file{counter:04d}.txt"))
1055 new.write(f"{counter}".encode())
1056 created.add(new)
1057 counter += 1
1059 found = set(ResourcePath.findFileResources([root]))
1060 self.assertEqual(len(found), n_dir1 + n_dir2)
1061 self.assertEqual(found, created)
1063 # Again with grouping.
1064 # (mypy gets upset not knowing which of the two options is being
1065 # returned so add useless instance check).
1066 found_list = [
1067 list(group)
1068 for group in ResourcePath.findFileResources([root], grouped=True)
1069 if not isinstance(group, ResourcePath) # For mypy.
1070 ]
1071 self.assertEqual(len(found_list), 2)
1072 self.assertEqual(len(found_list[0]), n_dir1)
1073 self.assertEqual(len(found_list[1]), n_dir2)
1075 def test_temporary(self) -> None:
1076 prefix = self.tmpdir.join("tmp", forceDirectory=True)
1077 with ResourcePath.temporary_uri(prefix=prefix, suffix=".json") as tmp:
1078 self.assertEqual(tmp.getExtension(), ".json", f"uri: {tmp}")
1079 self.assertTrue(tmp.isabs(), f"uri: {tmp}")
1080 self.assertFalse(tmp.exists(), f"uri: {tmp}")
1081 tmp.write(b"abcd")
1082 self.assertTrue(tmp.exists(), f"uri: {tmp}")
1083 self.assertTrue(tmp.isTemporary)
1084 self.assertFalse(tmp.exists(), f"uri: {tmp}")
1086 tmpdir = ResourcePath(self.tmpdir, forceDirectory=True)
1087 with ResourcePath.temporary_uri(prefix=tmpdir) as tmp:
1088 # Use a specified tmpdir and check it is okay for the file
1089 # to not be created.
1090 self.assertFalse(tmp.getExtension())
1091 self.assertFalse(tmp.exists(), f"uri: {tmp}")
1092 self.assertEqual(tmp.scheme, self.scheme)
1093 self.assertTrue(tmp.isTemporary)
1094 self.assertTrue(tmpdir.exists(), f"uri: {tmpdir} still exists")
1096 # Fake a directory suffix.
1097 with self.assertRaises(NotImplementedError):
1098 with ResourcePath.temporary_uri(prefix=self.root_uri, suffix="xxx/") as tmp:
1099 pass
1101 @unittest.skipIf(fsspec is None, "fsspec is not available.")
1102 def test_fsspec(self) -> None:
1103 """Simple read of a file."""
1104 uri = self.tmpdir.join("test.txt")
1105 self.assertFalse(uri.exists(), f"{uri} should not exist")
1106 self.assertTrue(uri.path.endswith("test.txt"))
1108 content = "abcdefghijklmnopqrstuv\n"
1109 uri.write(content.encode())
1111 try:
1112 fs, path = uri.to_fsspec()
1113 except NotImplementedError as e:
1114 raise unittest.SkipTest(str(e)) from e
1115 except ImportError as e:
1116 # HttpResourcePath.to_fsspec() raises if support
1117 # of fsspec for webDAV back ends is disabled.
1118 raise unittest.SkipTest(str(e)) from e
1119 with fs.open(path, "r") as fd:
1120 as_read = fd.read()
1121 self.assertEqual(as_read, content)
1123 def test_open(self) -> None:
1124 tmpdir = ResourcePath(self.tmpdir, forceDirectory=True)
1125 with ResourcePath.temporary_uri(prefix=tmpdir, suffix=".txt") as tmp:
1126 _check_open(self, tmp, mode_suffixes=("", "t"))
1127 _check_open(self, tmp, mode_suffixes=("t",), encoding="utf-16")
1128 _check_open(self, tmp, mode_suffixes=("t",), prefer_file_temporary=True)
1129 _check_open(self, tmp, mode_suffixes=("t",), encoding="utf-16", prefer_file_temporary=True)
1130 with ResourcePath.temporary_uri(prefix=tmpdir, suffix=".dat") as tmp:
1131 _check_open(self, tmp, mode_suffixes=("b",))
1132 _check_open(self, tmp, mode_suffixes=("b",), prefer_file_temporary=True)
1133 # An encoding is documented as being ignored for binary IO, so
1134 # supplying one must not change the content or raise.
1135 _check_open(self, tmp, mode_suffixes=("b",), encoding="utf-16")
1136 _check_open(self, tmp, mode_suffixes=("b",), encoding="utf-16", prefer_file_temporary=True)
1138 with self.assertRaises(IsADirectoryError):
1139 with self.root_uri.open():
1140 pass
1142 def test_mexists(self) -> None:
1143 root = self.tmpdir.join("mexists/")
1145 # A file that is not there.
1146 file = root.join("config/basic/butler.yaml")
1148 # Create some files. Most schemes the code paths do not change for 10
1149 # vs 1000 files but in some schemes it does.
1150 expected_files = [f"dir1/f{n}.yaml" for n in range(self.n_mremove_files)]
1151 expected_uris = [root.join(f) for f in expected_files]
1152 for uri in expected_uris:
1153 uri.write(b"")
1154 self.assertTrue(uri.exists())
1155 expected_uris.append(file)
1157 # Force to run with fewer workers than there are files.
1158 multi = ResourcePath.mexists(expected_uris, num_workers=3)
1160 for uri, is_there in multi.items():
1161 if uri == file:
1162 self.assertFalse(is_there)
1163 else:
1164 self.assertTrue(is_there)
1166 # Clean up. Unfortunately POSIX raises a FileNotFoundError but
1167 # S3 boto does not complain if there is no key.
1168 ResourcePath.mremove(expected_uris, do_raise=False)
1170 # Check they were really removed.
1171 multi = ResourcePath.mexists(expected_uris, num_workers=3)
1172 for uri, is_there in multi.items():
1173 self.assertFalse(is_there)
1175 # Clean up a subset of files that are already gone, but this can
1176 # trigger a different code path.
1177 ResourcePath.mremove(expected_uris[:5], do_raise=False)
1180def make_remote_test_uri(path: str, *, forceDirectory: bool = True) -> ResourcePath:
1181 """Return a ``remote-test`` URI corresponding to a local file system path.
1183 Parameters
1184 ----------
1185 path : `str`
1186 Local file system path to wrap.
1187 forceDirectory : `bool`, optional
1188 If `True`, the returned URI is treated as a directory.
1190 Returns
1191 -------
1192 uri : `lsst.resources.ResourcePath`
1193 URI using the ``remote-test`` scheme, which reports itself as not
1194 local while being backed by ``path``.
1196 Notes
1197 -----
1198 `~lsst.resources.ResourcePath` percent-encodes a path only when it is given
1199 a schemeless path; a string that already includes a scheme is used
1200 verbatim. The URI is therefore built from the encoded path of the
1201 equivalent ``file`` URI. Interpolating ``path`` directly would leave any
1202 space in the path unencoded, giving a malformed URI that.
1204 When a ``file`` URI is constructed only the ``path`` is copied to the
1205 new test URI. Any fragments are dropped (a ``#`` in the final component of
1206 the URI is always treated as a fragment by package convention).
1207 """
1208 file_uri = ResourcePath(path, forceDirectory=forceDirectory, forceAbsolute=True)
1209 return ResourcePath(f"remote-test://localhost{file_uri.path}", forceDirectory=forceDirectory)