Demonstrate python/cpython#127012

This adds an in-memory finder, loader, and traversable implementation,
which allows the `Traversable` protocol and concrete methods to be tested.

This additional infrastructure demonstrates python/cpython#127012,
but also highlights that the `Traversable.joinpath()` concrete method
raises `TraversalError` which is not getting caught in several places.

Author: kurtmckee

importlib_resources/tests/test_functional.py

@@ -8,10 +8,6 @@
 
 from . import util
 
-# Since the functional API forwards to Traversable, we only test
-# filesystem resources here -- not zip files, namespace packages etc.
-# We do test for two kinds of Anchor, though.
-
 
 class StringAnchorMixin:
     anchor01 = 'data01'
@@ -28,7 +24,7 @@ def anchor02(self):
         return importlib.import_module('data02')
 
 
-class FunctionalAPIBase(util.DiskSetup):
+class FunctionalAPIBase:
     def setUp(self):
         super().setUp()
         self.load_fixture('data02')
@@ -245,17 +241,28 @@ def test_text_errors(self):
                     )
 
 
-class FunctionalAPITest_StringAnchor(
+class FunctionalAPITest_StringAnchor_Disk(
     StringAnchorMixin,
     FunctionalAPIBase,
+    util.DiskSetup,
     unittest.TestCase,
 ):
     pass
 
 
-class FunctionalAPITest_ModuleAnchor(
+class FunctionalAPITest_ModuleAnchor_Disk(
     ModuleAnchorMixin,
     FunctionalAPIBase,
+    util.DiskSetup,
+    unittest.TestCase,
+):
+    pass
+
+
+class FunctionalAPITest_StringAnchor_Memory(
+    StringAnchorMixin,
+    FunctionalAPIBase,
+    util.MemorySetup,
     unittest.TestCase,
 ):
     pass

importlib_resources/tests/test_util.py

@@ -0,0 +1,35 @@
+import unittest
+
+from .util import Traversable, MemorySetup
+
+
+class TestMemoryTraversableImplementation(unittest.TestCase):
+    def test_concrete_methods_are_not_overridden(self):
+        """`MemoryTraversable` must not override `Traversable` concrete methods.
+
+        This test is not an attempt to enforce a particular `Traversable` protocol;
+        it merely catches changes in the `Traversable` abstract/concrete methods
+        that have not been mirrored in the `MemoryTraversable` subclass.
+        """
+
+        traversable_concrete_methods = {
+            method
+            for method, value in Traversable.__dict__.items()
+            if callable(value) and method not in Traversable.__abstractmethods__
+        }
+        memory_traversable_concrete_methods = {
+            method
+            for method, value in MemorySetup.MemoryTraversable.__dict__.items()
+            if callable(value) and not method.startswith("__")
+        }
+        overridden_methods = (
+            memory_traversable_concrete_methods & traversable_concrete_methods
+        )
+
+        if overridden_methods:
+            raise AssertionError(
+                "MemorySetup.MemoryTraversable overrides Traversable concrete methods, "
+                "which may mask problems in the Traversable protocol. "
+                "Please remove the following methods in MemoryTraversable: "
+                + ", ".join(overridden_methods)
+            )

importlib_resources/tests/util.py

@@ -1,12 +1,13 @@
 import abc
+import functools
 import importlib
 import io
 import sys
 import types
 import pathlib
 import contextlib
 
-from ..abc import ResourceReader
+from ..abc import ResourceReader, TraversableResources, Traversable
 from .compat.py39 import import_helper, os_helper
 from . import zip as zip_
 from . import _path
@@ -202,5 +203,107 @@ def tree_on_path(self, spec):
         self.fixtures.enter_context(import_helper.DirsOnSysPath(temp_dir))
 
 
+class MemorySetup(ModuleSetup):
+    """Support loading a module in memory."""
+
+    MODULE = 'data01'
+
+    def load_fixture(self, module):
+        self.fixtures.enter_context(self.augment_sys_metapath(module))
+        return importlib.import_module(module)
+
+    @contextlib.contextmanager
+    def augment_sys_metapath(self, module):
+        finder_instance = self.MemoryFinder(module)
+        sys.meta_path.append(finder_instance)
+        yield
+        sys.meta_path.remove(finder_instance)
+
+    class MemoryFinder(importlib.abc.MetaPathFinder):
+        def __init__(self, module):
+            self._module = module
+
+        def find_spec(self, fullname, path, target=None):
+            if fullname != self._module:
+                return None
+
+            return importlib.machinery.ModuleSpec(
+                name=fullname,
+                loader=MemorySetup.MemoryLoader(self._module),
+                is_package=True,
+            )
+
+    class MemoryLoader(importlib.abc.Loader):
+        def __init__(self, module):
+            self._module = module
+
+        def exec_module(self, module):
+            pass
+
+        def get_resource_reader(self, fullname):
+            return MemorySetup.MemoryTraversableResources(self._module, fullname)
+
+    class MemoryTraversableResources(TraversableResources):
+        def __init__(self, module, fullname):
+            self._module = module
+            self._fullname = fullname
+
+        def files(self):
+            return MemorySetup.MemoryTraversable(self._module, self._fullname)
+
+    class MemoryTraversable(Traversable):
+        """Implement only the abstract methods of `Traversable`.
+
+        Besides `.__init__()`, no other methods may be implemented or overridden.
+        This is critical for validating the concrete `Traversable` implementations.
+        """
+
+        def __init__(self, module, fullname):
+            self._module = module
+            self._fullname = fullname
+
+        def iterdir(self):
+            path = pathlib.PurePosixPath(self._fullname)
+            directory = functools.reduce(lambda d, p: d[p], path.parts, fixtures)
+            if not isinstance(directory, dict):
+                # Filesystem openers raise OSError, and that exception is mirrored here.
+                raise OSError(f"{self._fullname} is not a directory")
+            for path in directory:
+                yield MemorySetup.MemoryTraversable(
+                    self._module, f"{self._fullname}/{path}"
+                )
+
+        def is_dir(self) -> bool:
+            path = pathlib.PurePosixPath(self._fullname)
+            # Fully traverse the `fixtures` dictionary.
+            # This should be wrapped in a `try/except KeyError`
+            # but it is not currently needed, and lowers the code coverage numbers.
+            directory = functools.reduce(lambda d, p: d[p], path.parts, fixtures)
+            return isinstance(directory, dict)
+
+        def is_file(self) -> bool:
+            path = pathlib.PurePosixPath(self._fullname)
+            directory = functools.reduce(lambda d, p: d[p], path.parts, fixtures)
+            return not isinstance(directory, dict)
+
+        def open(self, mode='r', encoding=None, errors=None, *_, **__):
+            path = pathlib.PurePosixPath(self._fullname)
+            contents = functools.reduce(lambda d, p: d[p], path.parts, fixtures)
+            if isinstance(contents, dict):
+                # Filesystem openers raise OSError when attempting to open a directory,
+                # and that exception is mirrored here.
+                raise OSError(f"{self._fullname} is a directory")
+            if isinstance(contents, str):
+                contents = contents.encode("utf-8")
+            result = io.BytesIO(contents)
+            if "b" in mode:
+                return result
+            return io.TextIOWrapper(result, encoding=encoding, errors=errors)
+
+        @property
+        def name(self):
+            return pathlib.PurePosixPath(self._fullname).name
+
+
 class CommonTests(DiskSetup, CommonTestsBase):
     pass