Applies changes from importlib_resources 6.5.2.
"""
from ._common import (
+ Anchor,
+ Package,
as_file,
files,
- Package,
- Anchor,
)
-
from ._functional import (
contents,
is_resource,
read_binary,
read_text,
)
-
from .abc import ResourceReader
-
__all__ = [
'Package',
'Anchor',
-import os
-import pathlib
-import tempfile
-import functools
import contextlib
-import types
+import functools
import importlib
import inspect
import itertools
+import os
+import pathlib
+import tempfile
+import types
+from typing import cast, Optional, Union
-from typing import Union, Optional, cast
from .abc import ResourceReader, Traversable
Package = Union[types.ModuleType, str]
# zipimport.zipimporter does not support weak references, resulting in a
# TypeError. That seems terrible.
spec = package.__spec__
- reader = getattr(spec.loader, 'get_resource_reader', None) # type: ignore[union-attr]
+ reader = getattr(spec.loader, "get_resource_reader", None) # type: ignore[union-attr]
if reader is None:
return None
return reader(spec.name) # type: ignore[union-attr]
@resolve.register
def _(cand: None) -> types.ModuleType:
- return resolve(_infer_caller().f_globals['__name__'])
+ return resolve(_infer_caller().f_globals["__name__"])
def _infer_caller():
return frame_info.filename == stack[0].filename
def is_wrapper(frame_info):
- return frame_info.function == 'wrapper'
+ return frame_info.function == "wrapper"
stack = inspect.stack()
not_this_file = itertools.filterfalse(is_this_file, stack)
@contextlib.contextmanager
def _tempfile(
reader,
- suffix='',
+ suffix="",
# gh-93353: Keep a reference to call os.remove() in late Python
# finalization.
*,
import warnings
-from ._common import files, as_file
-
+from ._common import as_file, files
+from .abc import TraversalError
_MISSING = object()
Otherwise returns ``False``.
"""
- return _get_resource(anchor, path_names).is_file()
+ try:
+ return _get_resource(anchor, path_names).is_file()
+ except TraversalError:
+ return False
def contents(anchor, *path_names):
import abc
-import io
import itertools
import os
import pathlib
-from typing import Any, BinaryIO, Iterable, Iterator, NoReturn, Text, Optional
-from typing import runtime_checkable, Protocol
-from typing import Union
-
+from typing import (
+ Any,
+ BinaryIO,
+ Iterable,
+ Iterator,
+ Literal,
+ NoReturn,
+ Optional,
+ Protocol,
+ Text,
+ TextIO,
+ Union,
+ overload,
+ runtime_checkable,
+)
StrPath = Union[str, os.PathLike[str]]
with self.open('rb') as strm:
return strm.read()
- def read_text(self, encoding: Optional[str] = None) -> str:
+ def read_text(
+ self, encoding: Optional[str] = None, errors: Optional[str] = None
+ ) -> str:
"""
Read contents of self as text
"""
- with self.open(encoding=encoding) as strm:
+ with self.open(encoding=encoding, errors=errors) as strm:
return strm.read()
@abc.abstractmethod
"""
return self.joinpath(child)
+ @overload
+ def open(self, mode: Literal['r'] = 'r', *args: Any, **kwargs: Any) -> TextIO: ...
+
+ @overload
+ def open(self, mode: Literal['rb'], *args: Any, **kwargs: Any) -> BinaryIO: ...
+
@abc.abstractmethod
- def open(self, mode='r', *args, **kwargs):
+ def open(
+ self, mode: str = 'r', *args: Any, **kwargs: Any
+ ) -> Union[TextIO, BinaryIO]:
"""
mode may be 'r' or 'rb' to open as text or binary. Return a handle
suitable for reading (same as pathlib.Path.open).
def files(self) -> "Traversable":
"""Return a Traversable object for the loaded package."""
- def open_resource(self, resource: StrPath) -> io.BufferedReader:
+ def open_resource(self, resource: StrPath) -> BinaryIO:
return self.files().joinpath(resource).open('rb')
def resource_path(self, resource: Any) -> NoReturn:
import collections
import contextlib
import itertools
-import pathlib
import operator
+import pathlib
import re
import warnings
import zipfile
from collections.abc import Iterator
from . import abc
-
from ._itertools import only
-import pathlib
import functools
-
-from typing import Dict, Union
-from typing import runtime_checkable
-from typing import Protocol
-
+import pathlib
+from typing import Dict, Protocol, Union, runtime_checkable
####
# from jaraco.path 3.7.1
+import importlib.resources as resources
import io
import unittest
-
-from importlib import resources
-
from importlib.resources._adapters import (
CompatibilityFiles,
wrap_spec,
+import importlib.resources as resources
import unittest
-from importlib import resources
from . import util
-import unittest
import contextlib
+import importlib.resources as resources
import pathlib
+import unittest
+from importlib.resources import abc
+from importlib.resources.abc import ResourceReader, TraversableResources
from test.support import os_helper
-from importlib import resources
-from importlib.resources import abc
-from importlib.resources.abc import TraversableResources, ResourceReader
from . import util
+import contextlib
+import importlib
+import importlib.resources as resources
import pathlib
import py_compile
import textwrap
import unittest
import warnings
-import importlib
-import contextlib
-
-from importlib import resources
from importlib.resources.abc import Traversable
+
+from test.support import import_helper, os_helper
+
from . import util
-from test.support import os_helper, import_helper
@contextlib.contextmanager
to cause the ``PathEntryFinder`` to be called when searching
for packages. In that case, resources should still be loadable.
"""
- import namespacedata01
+ import namespacedata01 # type: ignore[import-not-found]
namespacedata01.__path__.append(
'__editable__.sample_namespace-1.0.finder.__path_hook__'
sources = pathlib.Path(resources.__file__).parent
for source_path in sources.glob('**/*.py'):
- c_path = c_resources.joinpath(source_path.relative_to(sources)).with_suffix('.pyc')
+ c_path = c_resources.joinpath(source_path.relative_to(sources)).with_suffix(
+ '.pyc'
+ )
py_compile.compile(source_path, c_path)
self.fixtures.enter_context(import_helper.DirsOnSysPath(bin_site))
-import unittest
-import os
import importlib
+import importlib.resources as resources
+import os
+import unittest
from test.support import warnings_helper
-from importlib import resources
-
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'
return importlib.import_module('data02')
-class FunctionalAPIBase(util.DiskSetup):
+class FunctionalAPIBase:
def setUp(self):
super().setUp()
self.load_fixture('data02')
with self.subTest(path_parts=path_parts):
yield path_parts
+ def assertEndsWith(self, string, suffix):
+ """Assert that `string` ends with `suffix`.
+
+ Used to ignore an architecture-specific UTF-16 byte-order mark."""
+ self.assertEqual(string[-len(suffix) :], suffix)
+
def test_read_text(self):
self.assertEqual(
resources.read_text(self.anchor01, 'utf-8.file'),
# fail with PermissionError rather than IsADirectoryError
with self.assertRaises(OSError):
resources.read_text(self.anchor01)
- with self.assertRaises(OSError):
+ with self.assertRaises((OSError, resources.abc.TraversalError)):
resources.read_text(self.anchor01, 'no-such-file')
with self.assertRaises(UnicodeDecodeError):
resources.read_text(self.anchor01, 'utf-16.file')
# fail with PermissionError rather than IsADirectoryError
with self.assertRaises(OSError):
resources.open_text(self.anchor01)
- with self.assertRaises(OSError):
+ with self.assertRaises((OSError, resources.abc.TraversalError)):
resources.open_text(self.anchor01, 'no-such-file')
with resources.open_text(self.anchor01, 'utf-16.file') as f:
with self.assertRaises(UnicodeDecodeError):
set(c),
{'utf-8.file', 'utf-16.file', 'binary.file', 'subdirectory'},
)
- with self.assertRaises(OSError), warnings_helper.check_warnings((
- ".*contents.*",
- DeprecationWarning,
- )):
+ with (
+ self.assertRaises(OSError),
+ warnings_helper.check_warnings((
+ ".*contents.*",
+ DeprecationWarning,
+ )),
+ ):
list(resources.contents(self.anchor01, 'utf-8.file'))
for path_parts in self._gen_resourcetxt_path_parts():
- with self.assertRaises(OSError), warnings_helper.check_warnings((
- ".*contents.*",
- DeprecationWarning,
- )):
+ with (
+ self.assertRaises((OSError, resources.abc.TraversalError)),
+ warnings_helper.check_warnings((
+ ".*contents.*",
+ DeprecationWarning,
+ )),
+ ):
list(resources.contents(self.anchor01, *path_parts))
with warnings_helper.check_warnings((".*contents.*", DeprecationWarning)):
c = resources.contents(self.anchor01, 'subdirectory')
)
-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
+import importlib.resources as resources
import unittest
-from importlib import resources
from . import util
+import importlib.resources as resources
import io
import pathlib
import unittest
-from importlib import resources
from . import util
target = resources.files(self.data) / 'utf-8.file'
with resources.as_file(target) as path:
self.assertIsInstance(path, pathlib.Path)
- self.assertEndsWith(path.name, "utf-8.file")
+ self.assertTrue(path.name.endswith("utf-8.file"), repr(path))
self.assertEqual('Hello, UTF-8 world!\n', path.read_text(encoding='utf-8'))
+import importlib.resources as resources
import unittest
-
-from importlib import import_module, resources
+from importlib import import_module
from . import util
import os.path
import pathlib
import unittest
-
from importlib import import_module
-from importlib.readers import MultiplexedPath, NamespaceReader
+from importlib.resources.readers import MultiplexedPath, NamespaceReader
from . import util
+import importlib.resources as resources
import unittest
+from importlib import import_module
from . import util
-from importlib import resources, import_module
class ResourceTests:
--- /dev/null
+import unittest
+
+from .util import MemorySetup, Traversable
+
+
+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
+ )
+
+ assert not overridden_methods
import abc
+import contextlib
+import functools
import importlib
import io
+import pathlib
import sys
import types
-import pathlib
-import contextlib
+from importlib.machinery import ModuleSpec
+from importlib.resources.abc import ResourceReader, Traversable, TraversableResources
-from importlib.resources.abc import ResourceReader
from test.support import import_helper, os_helper
-from . import zip as zip_
-from . import _path
-
-from importlib.machinery import ModuleSpec
+from . import _path
+from . import zip as zip_
class Reader(ResourceReader):
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 _resolve(self):
+ """
+ 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.
+ """
+ path = pathlib.PurePosixPath(self._fullname)
+ return functools.reduce(lambda d, p: d[p], path.parts, fixtures)
+
+ def iterdir(self):
+ directory = self._resolve()
+ 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:
+ return isinstance(self._resolve(), dict)
+
+ def is_file(self) -> bool:
+ return not self.is_dir()
+
+ def open(self, mode='r', encoding=None, errors=None, *_, **__):
+ contents = self._resolve()
+ 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
--- /dev/null
+``importlib.abc.Traversable.read_text`` now allows/solicits an
+``errors`` parameter.
projects / thirdparty / Python / cpython.git / commitdiff
