gh-97930: Apply changes from importlib_resources 5.12. (GH-102010)

Author: jaraco

Lib/importlib/resources/_adapters.py

@@ -34,9 +34,7 @@ def _io_wrapper(file, mode='r', *args, **kwargs):
         return TextIOWrapper(file, *args, **kwargs)
     elif mode == 'rb':
         return file
-    raise ValueError(
-        f"Invalid mode value '{mode}', only 'r' and 'rb' are supported"
-    )
+    raise ValueError(f"Invalid mode value '{mode}', only 'r' and 'rb' are supported")
 
 
 class CompatibilityFiles:

Lib/importlib/resources/_itertools.py

@@ -1,35 +1,38 @@
-from itertools import filterfalse
+# from more_itertools 9.0
+def only(iterable, default=None, too_long=None):
+    """If *iterable* has only one item, return it.
+    If it has zero items, return *default*.
+    If it has more than one item, raise the exception given by *too_long*,
+    which is ``ValueError`` by default.
+    >>> only([], default='missing')
+    'missing'
+    >>> only([1])
+    1
+    >>> only([1, 2])  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+    ...
+    ValueError: Expected exactly one item in iterable, but got 1, 2,
+     and perhaps more.'
+    >>> only([1, 2], too_long=TypeError)  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+    ...
+    TypeError
+    Note that :func:`only` attempts to advance *iterable* twice to ensure there
+    is only one item.  See :func:`spy` or :func:`peekable` to check
+    iterable contents less destructively.
+    """
+    it = iter(iterable)
+    first_value = next(it, default)
 
-from typing import (
-    Callable,
-    Iterable,
-    Iterator,
-    Optional,
-    Set,
-    TypeVar,
-    Union,
-)
-
-# Type and type variable definitions
-_T = TypeVar('_T')
-_U = TypeVar('_U')
-
-
-def unique_everseen(
-    iterable: Iterable[_T], key: Optional[Callable[[_T], _U]] = None
-) -> Iterator[_T]:
-    "List unique elements, preserving order. Remember all elements ever seen."
-    # unique_everseen('AAAABBBCCDAABBB') --> A B C D
-    # unique_everseen('ABBCcAD', str.lower) --> A B C D
-    seen: Set[Union[_T, _U]] = set()
-    seen_add = seen.add
-    if key is None:
-        for element in filterfalse(seen.__contains__, iterable):
-            seen_add(element)
-            yield element
+    try:
+        second_value = next(it)
+    except StopIteration:
+        pass
     else:
-        for element in iterable:
-            k = key(element)
-            if k not in seen:
-                seen_add(k)
-                yield element
+        msg = (
+            'Expected exactly one item in iterable, but got {!r}, {!r}, '
+            'and perhaps more.'.format(first_value, second_value)
+        )
+        raise too_long or ValueError(msg)
+
+    return first_value

Lib/importlib/resources/readers.py

@@ -1,11 +1,12 @@
 import collections
-import operator
+import itertools
 import pathlib
+import operator
 import zipfile
 
 from . import abc
 
-from ._itertools import unique_everseen
+from ._itertools import only
 
 
 def remove_duplicates(items):
@@ -41,8 +42,10 @@ def open_resource(self, resource):
             raise FileNotFoundError(exc.args[0])
 
     def is_resource(self, path):
-        # workaround for `zipfile.Path.is_file` returning true
-        # for non-existent paths.
+        """
+        Workaround for `zipfile.Path.is_file` returning true
+        for non-existent paths.
+        """
         target = self.files().joinpath(path)
         return target.is_file() and target.exists()
 
@@ -67,8 +70,10 @@ def __init__(self, *paths):
             raise NotADirectoryError('MultiplexedPath only supports directories')
 
     def iterdir(self):
-        files = (file for path in self._paths for file in path.iterdir())
-        return unique_everseen(files, key=operator.attrgetter('name'))
+        children = (child for path in self._paths for child in path.iterdir())
+        by_name = operator.attrgetter('name')
+        groups = itertools.groupby(sorted(children, key=by_name), key=by_name)
+        return map(self._follow, (locs for name, locs in groups))
 
     def read_bytes(self):
         raise FileNotFoundError(f'{self} is not a file')
@@ -90,6 +95,25 @@ def joinpath(self, *descendants):
             # Just return something that will not exist.
             return self._paths[0].joinpath(*descendants)
 
+    @classmethod
+    def _follow(cls, children):
+        """
+        Construct a MultiplexedPath if needed.
+
+        If children contains a sole element, return it.
+        Otherwise, return a MultiplexedPath of the items.
+        Unless one of the items is not a Directory, then return the first.
+        """
+        subdirs, one_dir, one_file = itertools.tee(children, 3)
+
+        try:
+            return only(one_dir)
+        except ValueError:
+            try:
+                return cls(*subdirs)
+            except NotADirectoryError:
+                return next(one_file)
+
     def open(self, *args, **kwargs):
         raise FileNotFoundError(f'{self} is not a file')
 

Lib/test/test_importlib/resources/_path.py

@@ -1,12 +1,16 @@
 import pathlib
 import functools
 
+from typing import Dict, Union
+
 
 ####
-# from jaraco.path 3.4
+# from jaraco.path 3.4.1
+
+FilesSpec = Dict[str, Union[str, bytes, 'FilesSpec']]  # type: ignore
 
 
-def build(spec, prefix=pathlib.Path()):
+def build(spec: FilesSpec, prefix=pathlib.Path()):
     """
     Build a set of files/directories, as described by the spec.
 
@@ -23,15 +27,17 @@ def build(spec, prefix=pathlib.Path()):
     ...         "baz.py": "# Some code",
     ...     }
     ... }
-    >>> tmpdir = getfixture('tmpdir')
-    >>> build(spec, tmpdir)
+    >>> target = getfixture('tmp_path')
+    >>> build(spec, target)
+    >>> target.joinpath('foo/baz.py').read_text(encoding='utf-8')
+    '# Some code'
     """
     for name, contents in spec.items():
         create(contents, pathlib.Path(prefix) / name)
 
 
 @functools.singledispatch
-def create(content, path):
+def create(content: Union[str, bytes, FilesSpec], path):
     path.mkdir(exist_ok=True)
     build(content, prefix=path)  # type: ignore
 
@@ -43,7 +49,7 @@ def _(content: bytes, path):
 
 @create.register
 def _(content: str, path):
-    path.write_text(content)
+    path.write_text(content, encoding='utf-8')
 
 
 # end from jaraco.path

Lib/test/test_importlib/resources/data02/subdirectory/subsubdir/resource.txt

@@ -0,0 +1 @@
+a resource

Lib/test/test_importlib/resources/test_compatibilty_files.py

@@ -64,11 +64,13 @@ def test_orphan_path_name(self):
 
     def test_spec_path_open(self):
         self.assertEqual(self.files.read_bytes(), b'Hello, world!')
-        self.assertEqual(self.files.read_text(), 'Hello, world!')
+        self.assertEqual(self.files.read_text(encoding='utf-8'), 'Hello, world!')
 
     def test_child_path_open(self):
         self.assertEqual((self.files / 'a').read_bytes(), b'Hello, world!')
-        self.assertEqual((self.files / 'a').read_text(), 'Hello, world!')
+        self.assertEqual(
+            (self.files / 'a').read_text(encoding='utf-8'), 'Hello, world!'
+        )
 
     def test_orphan_path_open(self):
         with self.assertRaises(FileNotFoundError):

Lib/test/test_importlib/resources/test_custom.py

@@ -0,0 +1,46 @@
+import unittest
+import contextlib
+import pathlib
+
+from test.support import os_helper
+
+from importlib import resources
+from importlib.resources.abc import TraversableResources, ResourceReader
+from . import util
+
+
+class SimpleLoader:
+    """
+    A simple loader that only implements a resource reader.
+    """
+
+    def __init__(self, reader: ResourceReader):
+        self.reader = reader
+
+    def get_resource_reader(self, package):
+        return self.reader
+
+
+class MagicResources(TraversableResources):
+    """
+    Magically returns the resources at path.
+    """
+
+    def __init__(self, path: pathlib.Path):
+        self.path = path
+
+    def files(self):
+        return self.path
+
+
+class CustomTraversableResourcesTests(unittest.TestCase):
+    def setUp(self):
+        self.fixtures = contextlib.ExitStack()
+        self.addCleanup(self.fixtures.close)
+
+    def test_custom_loader(self):
+        temp_dir = self.fixtures.enter_context(os_helper.temp_dir())
+        loader = SimpleLoader(MagicResources(temp_dir))
+        pkg = util.create_package_from_loader(loader)
+        files = resources.files(pkg)
+        assert files is temp_dir

Lib/test/test_importlib/resources/test_files.py

@@ -85,7 +85,7 @@ def test_module_resources(self):
         _path.build(spec, self.site_dir)
         import mod
 
-        actual = resources.files(mod).joinpath('res.txt').read_text()
+        actual = resources.files(mod).joinpath('res.txt').read_text(encoding='utf-8')
         assert actual == spec['res.txt']
 
 
@@ -99,7 +99,7 @@ def test_implicit_files(self):
                 '__init__.py': textwrap.dedent(
                     """
                     import importlib.resources as res
-                    val = res.files().joinpath('res.txt').read_text()
+                    val = res.files().joinpath('res.txt').read_text(encoding='utf-8')
                     """
                 ),
                 'res.txt': 'resources are the best',

Lib/test/test_importlib/resources/test_open.py

@@ -15,7 +15,7 @@ def execute(self, package, path):
 class CommonTextTests(util.CommonTests, unittest.TestCase):
     def execute(self, package, path):
         target = resources.files(package).joinpath(path)
-        with target.open():
+        with target.open(encoding='utf-8'):
             pass
 
 
@@ -28,7 +28,7 @@ def test_open_binary(self):
 
     def test_open_text_default_encoding(self):
         target = resources.files(self.data) / 'utf-8.file'
-        with target.open() as fp:
+        with target.open(encoding='utf-8') as fp:
             result = fp.read()
             self.assertEqual(result, 'Hello, UTF-8 world!\n')
 
@@ -39,7 +39,9 @@ def test_open_text_given_encoding(self):
         self.assertEqual(result, 'Hello, UTF-16 world!\n')
 
     def test_open_text_with_errors(self):
-        # Raises UnicodeError without the 'errors' argument.
+        """
+        Raises UnicodeError without the 'errors' argument.
+        """
         target = resources.files(self.data) / 'utf-16.file'
         with target.open(encoding='utf-8', errors='strict') as fp:
             self.assertRaises(UnicodeError, fp.read)
@@ -54,11 +56,13 @@ def test_open_text_with_errors(self):
 
     def test_open_binary_FileNotFoundError(self):
         target = resources.files(self.data) / 'does-not-exist'
-        self.assertRaises(FileNotFoundError, target.open, 'rb')
+        with self.assertRaises(FileNotFoundError):
+            target.open('rb')
 
     def test_open_text_FileNotFoundError(self):
         target = resources.files(self.data) / 'does-not-exist'
-        self.assertRaises(FileNotFoundError, target.open)
+        with self.assertRaises(FileNotFoundError):
+            target.open(encoding='utf-8')
 
 
 class OpenDiskTests(OpenTests, unittest.TestCase):

Lib/test/test_importlib/resources/test_path.py

@@ -14,9 +14,12 @@ def execute(self, package, path):
 
 class PathTests:
     def test_reading(self):
-        # Path should be readable.
-        # Test also implicitly verifies the returned object is a pathlib.Path
-        # instance.
+        """
+        Path should be readable.
+
+        Test also implicitly verifies the returned object is a pathlib.Path
+        instance.
+        """
         target = resources.files(self.data) / 'utf-8.file'
         with resources.as_file(target) as path:
             self.assertTrue(path.name.endswith("utf-8.file"), repr(path))
@@ -51,8 +54,10 @@ def setUp(self):
 
 class PathZipTests(PathTests, util.ZipSetup, unittest.TestCase):
     def test_remove_in_context_manager(self):
-        # It is not an error if the file that was temporarily stashed on the
-        # file system is removed inside the `with` stanza.
+        """
+        It is not an error if the file that was temporarily stashed on the
+        file system is removed inside the `with` stanza.
+        """
         target = resources.files(self.data) / 'utf-8.file'
         with resources.as_file(target) as path:
             path.unlink()