bpo-44771: Apply changes from importlib_resources 5.2.1 (GH-27436)

* bpo-44771: Apply changes from importlib_resources@3b24bd6307

* Add blurb

* Exclude namespacedata01 from eol conversion.
This commit is contained in:
Jason R. Coombs 2021-07-29 21:05:05 -04:00 committed by GitHub
parent 851cca8c22
commit aaa83cdfab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 716 additions and 375 deletions

1
.gitattributes vendored
View File

@ -28,6 +28,7 @@ Lib/test/test_email/data/*.txt -text
Lib/test/xmltestdata/* -text Lib/test/xmltestdata/* -text
Lib/test/coding20731.py -text Lib/test/coding20731.py -text
Lib/test/test_importlib/data01/* -text Lib/test/test_importlib/data01/* -text
Lib/test/test_importlib/namespacedata01/* -text
# CRLF files # CRLF files
*.bat text eol=crlf *.bat text eol=crlf

View File

@ -1,4 +1,5 @@
from contextlib import suppress from contextlib import suppress
from io import TextIOWrapper
from . import abc from . import abc
@ -25,32 +26,119 @@ class TraversableResourcesLoader:
self.spec = spec self.spec = spec
def get_resource_reader(self, name): def get_resource_reader(self, name):
return DegenerateFiles(self.spec)._native() return CompatibilityFiles(self.spec)._native()
class DegenerateFiles: def _io_wrapper(file, mode='r', *args, **kwargs):
if mode == 'r':
return TextIOWrapper(file, *args, **kwargs)
elif mode == 'rb':
return file
raise ValueError(
"Invalid mode value '{}', only 'r' and 'rb' are supported".format(mode)
)
class CompatibilityFiles:
""" """
Adapter for an existing or non-existant resource reader Adapter for an existing or non-existant resource reader
to provide a degenerate .files(). to provide a compability .files().
""" """
class Path(abc.Traversable): class SpecPath(abc.Traversable):
"""
Path tied to a module spec.
Can be read and exposes the resource reader children.
"""
def __init__(self, spec, reader):
self._spec = spec
self._reader = reader
def iterdir(self):
if not self._reader:
return iter(())
return iter(
CompatibilityFiles.ChildPath(self._reader, path)
for path in self._reader.contents()
)
def is_file(self):
return False
is_dir = is_file
def joinpath(self, other):
if not self._reader:
return CompatibilityFiles.OrphanPath(other)
return CompatibilityFiles.ChildPath(self._reader, other)
@property
def name(self):
return self._spec.name
def open(self, mode='r', *args, **kwargs):
return _io_wrapper(self._reader.open_resource(None), mode, *args, **kwargs)
class ChildPath(abc.Traversable):
"""
Path tied to a resource reader child.
Can be read but doesn't expose any meaningfull children.
"""
def __init__(self, reader, name):
self._reader = reader
self._name = name
def iterdir(self): def iterdir(self):
return iter(()) return iter(())
def is_dir(self): def is_file(self):
return False return self._reader.is_resource(self.name)
is_file = exists = is_dir # type: ignore def is_dir(self):
return not self.is_file()
def joinpath(self, other): def joinpath(self, other):
return DegenerateFiles.Path() return CompatibilityFiles.OrphanPath(self.name, other)
@property
def name(self): def name(self):
return '' return self._name
def open(self): def open(self, mode='r', *args, **kwargs):
raise ValueError() return _io_wrapper(
self._reader.open_resource(self.name), mode, *args, **kwargs
)
class OrphanPath(abc.Traversable):
"""
Orphan path, not tied to a module spec or resource reader.
Can't be read and doesn't expose any meaningful children.
"""
def __init__(self, *path_parts):
if len(path_parts) < 1:
raise ValueError('Need at least one path part to construct a path')
self._path = path_parts
def iterdir(self):
return iter(())
def is_file(self):
return False
is_dir = is_file
def joinpath(self, other):
return CompatibilityFiles.OrphanPath(*self._path, other)
@property
def name(self):
return self._path[-1]
def open(self, mode='r', *args, **kwargs):
raise FileNotFoundError("Can't open orphan path")
def __init__(self, spec): def __init__(self, spec):
self.spec = spec self.spec = spec
@ -71,7 +159,7 @@ class DegenerateFiles:
return getattr(self._reader, attr) return getattr(self._reader, attr)
def files(self): def files(self):
return DegenerateFiles.Path() return CompatibilityFiles.SpecPath(self.spec, self._reader)
def wrap_spec(package): def wrap_spec(package):

View File

@ -12,6 +12,7 @@ from .abc import ResourceReader, Traversable
from ._adapters import wrap_spec from ._adapters import wrap_spec
Package = Union[types.ModuleType, str] Package = Union[types.ModuleType, str]
Resource = Union[str, os.PathLike]
def files(package): def files(package):
@ -93,7 +94,7 @@ def _tempfile(reader, suffix=''):
finally: finally:
try: try:
os.remove(raw_path) os.remove(raw_path)
except FileNotFoundError: except (FileNotFoundError, PermissionError):
pass pass

View File

@ -0,0 +1,19 @@
from itertools import filterfalse
def unique_everseen(iterable, key=None):
"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()
seen_add = seen.add
if key is None:
for element in filterfalse(seen.__contains__, iterable):
seen_add(element)
yield element
else:
for element in iterable:
k = key(element)
if k not in seen:
seen_add(k)
yield element

84
Lib/importlib/_legacy.py Normal file
View File

@ -0,0 +1,84 @@
import os
import pathlib
import types
from typing import Union, Iterable, ContextManager, BinaryIO, TextIO
from . import _common
Package = Union[types.ModuleType, str]
Resource = Union[str, os.PathLike]
def open_binary(package: Package, resource: Resource) -> BinaryIO:
"""Return a file-like object opened for binary reading of the resource."""
return (_common.files(package) / _common.normalize_path(resource)).open('rb')
def read_binary(package: Package, resource: Resource) -> bytes:
"""Return the binary contents of the resource."""
return (_common.files(package) / _common.normalize_path(resource)).read_bytes()
def open_text(
package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict',
) -> TextIO:
"""Return a file-like object opened for text reading of the resource."""
return (_common.files(package) / _common.normalize_path(resource)).open(
'r', encoding=encoding, errors=errors
)
def read_text(
package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict',
) -> str:
"""Return the decoded string of the resource.
The decoding-related arguments have the same semantics as those of
bytes.decode().
"""
with open_text(package, resource, encoding, errors) as fp:
return fp.read()
def contents(package: Package) -> Iterable[str]:
"""Return an iterable of entries in `package`.
Note that not all entries are resources. Specifically, directories are
not considered resources. Use `is_resource()` on each entry returned here
to check if it is a resource or not.
"""
return [path.name for path in _common.files(package).iterdir()]
def is_resource(package: Package, name: str) -> bool:
"""True if `name` is a resource inside `package`.
Directories are *not* resources.
"""
resource = _common.normalize_path(name)
return any(
traversable.name == resource and traversable.is_file()
for traversable in _common.files(package).iterdir()
)
def path(
package: Package,
resource: Resource,
) -> ContextManager[pathlib.Path]:
"""A context manager providing a file path object to the resource.
If the resource does not already exist on its own on the file system,
a temporary file will be created. If the file was created, the file
will be deleted upon exiting the context manager (no exception is
raised if the file was deleted prior to the context manager
exiting).
"""
return _common.as_file(_common.files(package) / _common.normalize_path(resource))

View File

@ -1,8 +1,12 @@
import collections import collections
import zipfile import operator
import pathlib import pathlib
import zipfile
from . import abc from . import abc
from ._itertools import unique_everseen
def remove_duplicates(items): def remove_duplicates(items):
return iter(collections.OrderedDict.fromkeys(items)) return iter(collections.OrderedDict.fromkeys(items))
@ -63,13 +67,8 @@ class MultiplexedPath(abc.Traversable):
raise NotADirectoryError('MultiplexedPath only supports directories') raise NotADirectoryError('MultiplexedPath only supports directories')
def iterdir(self): def iterdir(self):
visited = [] files = (file for path in self._paths for file in path.iterdir())
for path in self._paths: return unique_everseen(files, key=operator.attrgetter('name'))
for file in path.iterdir():
if file.name in visited:
continue
visited.append(file.name)
yield file
def read_bytes(self): def read_bytes(self):
raise FileNotFoundError(f'{self} is not a file') raise FileNotFoundError(f'{self} is not a file')

View File

@ -1,19 +1,23 @@
import os """Read resources contained within a package."""
import io
from . import _common from ._common import (
from ._common import as_file, files as_file,
from .abc import ResourceReader files,
from contextlib import suppress Package,
from importlib.abc import ResourceLoader Resource,
from importlib.machinery import ModuleSpec )
from io import BytesIO, TextIOWrapper
from pathlib import Path from ._legacy import (
from types import ModuleType contents,
from typing import ContextManager, Iterable, Union open_binary,
from typing import cast, BinaryIO, TextIO read_binary,
from collections.abc import Sequence open_text,
from functools import singledispatch read_text,
is_resource,
path,
)
from importlib.abc import ResourceReader
__all__ = [ __all__ = [
@ -30,155 +34,3 @@ __all__ = [
'read_binary', 'read_binary',
'read_text', 'read_text',
] ]
Package = Union[str, ModuleType]
Resource = Union[str, os.PathLike]
def open_binary(package: Package, resource: Resource) -> BinaryIO:
"""Return a file-like object opened for binary reading of the resource."""
resource = _common.normalize_path(resource)
package = _common.get_package(package)
reader = _common.get_resource_reader(package)
if reader is not None:
return reader.open_resource(resource)
spec = cast(ModuleSpec, package.__spec__)
# Using pathlib doesn't work well here due to the lack of 'strict'
# argument for pathlib.Path.resolve() prior to Python 3.6.
if spec.submodule_search_locations is not None:
paths = spec.submodule_search_locations
elif spec.origin is not None:
paths = [os.path.dirname(os.path.abspath(spec.origin))]
for package_path in paths:
full_path = os.path.join(package_path, resource)
try:
return open(full_path, mode='rb')
except OSError:
# Just assume the loader is a resource loader; all the relevant
# importlib.machinery loaders are and an AttributeError for
# get_data() will make it clear what is needed from the loader.
loader = cast(ResourceLoader, spec.loader)
data = None
if hasattr(spec.loader, 'get_data'):
with suppress(OSError):
data = loader.get_data(full_path)
if data is not None:
return BytesIO(data)
raise FileNotFoundError(f'{resource!r} resource not found in {spec.name!r}')
def open_text(
package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict',
) -> TextIO:
"""Return a file-like object opened for text reading of the resource."""
return TextIOWrapper(
open_binary(package, resource), encoding=encoding, errors=errors
)
def read_binary(package: Package, resource: Resource) -> bytes:
"""Return the binary contents of the resource."""
with open_binary(package, resource) as fp:
return fp.read()
def read_text(
package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict',
) -> str:
"""Return the decoded string of the resource.
The decoding-related arguments have the same semantics as those of
bytes.decode().
"""
with open_text(package, resource, encoding, errors) as fp:
return fp.read()
def path(
package: Package,
resource: Resource,
) -> 'ContextManager[Path]':
"""A context manager providing a file path object to the resource.
If the resource does not already exist on its own on the file system,
a temporary file will be created. If the file was created, the file
will be deleted upon exiting the context manager (no exception is
raised if the file was deleted prior to the context manager
exiting).
"""
reader = _common.get_resource_reader(_common.get_package(package))
return (
_path_from_reader(reader, _common.normalize_path(resource))
if reader
else _common.as_file(
_common.files(package).joinpath(_common.normalize_path(resource))
)
)
def _path_from_reader(reader, resource):
return _path_from_resource_path(reader, resource) or _path_from_open_resource(
reader, resource
)
def _path_from_resource_path(reader, resource):
with suppress(FileNotFoundError):
return Path(reader.resource_path(resource))
def _path_from_open_resource(reader, resource):
saved = io.BytesIO(reader.open_resource(resource).read())
return _common._tempfile(saved.read, suffix=resource)
def is_resource(package: Package, name: str) -> bool:
"""True if 'name' is a resource inside 'package'.
Directories are *not* resources.
"""
package = _common.get_package(package)
_common.normalize_path(name)
reader = _common.get_resource_reader(package)
if reader is not None:
return reader.is_resource(name)
package_contents = set(contents(package))
if name not in package_contents:
return False
return (_common.from_package(package) / name).is_file()
def contents(package: Package) -> Iterable[str]:
"""Return an iterable of entries in 'package'.
Note that not all entries are resources. Specifically, directories are
not considered resources. Use `is_resource()` on each entry returned here
to check if it is a resource or not.
"""
package = _common.get_package(package)
reader = _common.get_resource_reader(package)
if reader is not None:
return _ensure_sequence(reader.contents())
transversable = _common.from_package(package)
if transversable.is_dir():
return list(item.name for item in transversable.iterdir())
return []
@singledispatch
def _ensure_sequence(iterable):
return list(iterable)
@_ensure_sequence.register(Sequence)
def _(iterable):
return iterable

116
Lib/importlib/simple.py Normal file
View File

@ -0,0 +1,116 @@
"""
Interface adapters for low-level readers.
"""
import abc
import io
import itertools
from typing import BinaryIO, List
from .abc import Traversable, TraversableResources
class SimpleReader(abc.ABC):
"""
The minimum, low-level interface required from a resource
provider.
"""
@abc.abstractproperty
def package(self):
# type: () -> str
"""
The name of the package for which this reader loads resources.
"""
@abc.abstractmethod
def children(self):
# type: () -> List['SimpleReader']
"""
Obtain an iterable of SimpleReader for available
child containers (e.g. directories).
"""
@abc.abstractmethod
def resources(self):
# type: () -> List[str]
"""
Obtain available named resources for this virtual package.
"""
@abc.abstractmethod
def open_binary(self, resource):
# type: (str) -> BinaryIO
"""
Obtain a File-like for a named resource.
"""
@property
def name(self):
return self.package.split('.')[-1]
class ResourceHandle(Traversable):
"""
Handle to a named resource in a ResourceReader.
"""
def __init__(self, parent, name):
# type: (ResourceContainer, str) -> None
self.parent = parent
self.name = name # type: ignore
def is_file(self):
return True
def is_dir(self):
return False
def open(self, mode='r', *args, **kwargs):
stream = self.parent.reader.open_binary(self.name)
if 'b' not in mode:
stream = io.TextIOWrapper(*args, **kwargs)
return stream
def joinpath(self, name):
raise RuntimeError("Cannot traverse into a resource")
class ResourceContainer(Traversable):
"""
Traversable container for a package's resources via its reader.
"""
def __init__(self, reader):
# type: (SimpleReader) -> None
self.reader = reader
def is_dir(self):
return True
def is_file(self):
return False
def iterdir(self):
files = (ResourceHandle(self, name) for name in self.reader.resources)
dirs = map(ResourceContainer, self.reader.children())
return itertools.chain(files, dirs)
def open(self, *args, **kwargs):
raise IsADirectoryError()
def joinpath(self, name):
return next(
traversable for traversable in self.iterdir() if traversable.name == name
)
class TraversableReader(TraversableResources, SimpleReader):
"""
A TraversableResources based on SimpleReader. Resource providers
may derive from this class to provide the TraversableResources
interface by supplying the SimpleReader interface.
"""
def files(self):
return ResourceContainer(self)

View File

@ -0,0 +1,190 @@
import abc
import importlib
import io
import sys
import types
from pathlib import Path, PurePath
from .. import data01
from .. import zipdata01
from importlib.abc import ResourceReader
from test.support import import_helper
from importlib.machinery import ModuleSpec
class Reader(ResourceReader):
def __init__(self, **kwargs):
vars(self).update(kwargs)
def get_resource_reader(self, package):
return self
def open_resource(self, path):
self._path = path
if isinstance(self.file, Exception):
raise self.file
return self.file
def resource_path(self, path_):
self._path = path_
if isinstance(self.path, Exception):
raise self.path
return self.path
def is_resource(self, path_):
self._path = path_
if isinstance(self.path, Exception):
raise self.path
def part(entry):
return entry.split('/')
return any(
len(parts) == 1 and parts[0] == path_ for parts in map(part, self._contents)
)
def contents(self):
if isinstance(self.path, Exception):
raise self.path
yield from self._contents
def create_package_from_loader(loader, is_package=True):
name = 'testingpackage'
module = types.ModuleType(name)
spec = ModuleSpec(name, loader, origin='does-not-exist', is_package=is_package)
module.__spec__ = spec
module.__loader__ = loader
return module
def create_package(file=None, path=None, is_package=True, contents=()):
return create_package_from_loader(
Reader(file=file, path=path, _contents=contents),
is_package,
)
class CommonTests(metaclass=abc.ABCMeta):
"""
Tests shared by test_open, test_path, and test_read.
"""
@abc.abstractmethod
def execute(self, package, path):
"""
Call the pertinent legacy API function (e.g. open_text, path)
on package and path.
"""
def test_package_name(self):
# Passing in the package name should succeed.
self.execute(data01.__name__, 'utf-8.file')
def test_package_object(self):
# Passing in the package itself should succeed.
self.execute(data01, 'utf-8.file')
def test_string_path(self):
# Passing in a string for the path should succeed.
path = 'utf-8.file'
self.execute(data01, path)
def test_pathlib_path(self):
# Passing in a pathlib.PurePath object for the path should succeed.
path = PurePath('utf-8.file')
self.execute(data01, path)
def test_absolute_path(self):
# An absolute path is a ValueError.
path = Path(__file__)
full_path = path.parent / 'utf-8.file'
with self.assertRaises(ValueError):
self.execute(data01, full_path)
def test_relative_path(self):
# A reative path is a ValueError.
with self.assertRaises(ValueError):
self.execute(data01, '../data01/utf-8.file')
def test_importing_module_as_side_effect(self):
# The anchor package can already be imported.
del sys.modules[data01.__name__]
self.execute(data01.__name__, 'utf-8.file')
def test_non_package_by_name(self):
# The anchor package cannot be a module.
with self.assertRaises(TypeError):
self.execute(__name__, 'utf-8.file')
def test_non_package_by_package(self):
# The anchor package cannot be a module.
with self.assertRaises(TypeError):
module = sys.modules['test.test_importlib.resources.util']
self.execute(module, 'utf-8.file')
def test_missing_path(self):
# Attempting to open or read or request the path for a
# non-existent path should succeed if open_resource
# can return a viable data stream.
bytes_data = io.BytesIO(b'Hello, world!')
package = create_package(file=bytes_data, path=FileNotFoundError())
self.execute(package, 'utf-8.file')
self.assertEqual(package.__loader__._path, 'utf-8.file')
def test_extant_path(self):
# Attempting to open or read or request the path when the
# path does exist should still succeed. Does not assert
# anything about the result.
bytes_data = io.BytesIO(b'Hello, world!')
# any path that exists
path = __file__
package = create_package(file=bytes_data, path=path)
self.execute(package, 'utf-8.file')
self.assertEqual(package.__loader__._path, 'utf-8.file')
def test_useless_loader(self):
package = create_package(file=FileNotFoundError(), path=FileNotFoundError())
with self.assertRaises(FileNotFoundError):
self.execute(package, 'utf-8.file')
class ZipSetupBase:
ZIP_MODULE = None
@classmethod
def setUpClass(cls):
data_path = Path(cls.ZIP_MODULE.__file__)
data_dir = data_path.parent
cls._zip_path = str(data_dir / 'ziptestdata.zip')
sys.path.append(cls._zip_path)
cls.data = importlib.import_module('ziptestdata')
@classmethod
def tearDownClass(cls):
try:
sys.path.remove(cls._zip_path)
except ValueError:
pass
try:
del sys.path_importer_cache[cls._zip_path]
del sys.modules[cls.data.__name__]
except KeyError:
pass
try:
del cls.data
del cls._zip_path
except AttributeError:
pass
def setUp(self):
modules = import_helper.modules_setup()
self.addCleanup(import_helper.modules_cleanup, *modules)
class ZipSetup(ZipSetupBase):
ZIP_MODULE = zipdata01 # type: ignore

View File

@ -0,0 +1,102 @@
import io
import unittest
from importlib import resources
from importlib._adapters import (
CompatibilityFiles,
wrap_spec,
)
from .resources import util
class CompatibilityFilesTests(unittest.TestCase):
@property
def package(self):
bytes_data = io.BytesIO(b'Hello, world!')
return util.create_package(
file=bytes_data,
path='some_path',
contents=('a', 'b', 'c'),
)
@property
def files(self):
return resources.files(self.package)
def test_spec_path_iter(self):
self.assertEqual(
sorted(path.name for path in self.files.iterdir()),
['a', 'b', 'c'],
)
def test_child_path_iter(self):
self.assertEqual(list((self.files / 'a').iterdir()), [])
def test_orphan_path_iter(self):
self.assertEqual(list((self.files / 'a' / 'a').iterdir()), [])
self.assertEqual(list((self.files / 'a' / 'a' / 'a').iterdir()), [])
def test_spec_path_is(self):
self.assertFalse(self.files.is_file())
self.assertFalse(self.files.is_dir())
def test_child_path_is(self):
self.assertTrue((self.files / 'a').is_file())
self.assertFalse((self.files / 'a').is_dir())
def test_orphan_path_is(self):
self.assertFalse((self.files / 'a' / 'a').is_file())
self.assertFalse((self.files / 'a' / 'a').is_dir())
self.assertFalse((self.files / 'a' / 'a' / 'a').is_file())
self.assertFalse((self.files / 'a' / 'a' / 'a').is_dir())
def test_spec_path_name(self):
self.assertEqual(self.files.name, 'testingpackage')
def test_child_path_name(self):
self.assertEqual((self.files / 'a').name, 'a')
def test_orphan_path_name(self):
self.assertEqual((self.files / 'a' / 'b').name, 'b')
self.assertEqual((self.files / 'a' / 'b' / 'c').name, 'c')
def test_spec_path_open(self):
self.assertEqual(self.files.read_bytes(), b'Hello, world!')
self.assertEqual(self.files.read_text(), '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!')
def test_orphan_path_open(self):
with self.assertRaises(FileNotFoundError):
(self.files / 'a' / 'b').read_bytes()
with self.assertRaises(FileNotFoundError):
(self.files / 'a' / 'b' / 'c').read_bytes()
def test_open_invalid_mode(self):
with self.assertRaises(ValueError):
self.files.open('0')
def test_orphan_path_invalid(self):
with self.assertRaises(ValueError):
CompatibilityFiles.OrphanPath()
def test_wrap_spec(self):
spec = wrap_spec(self.package)
self.assertIsInstance(spec.loader.get_resource_reader(None), CompatibilityFiles)
class CompatibilityFilesNoReaderTests(unittest.TestCase):
@property
def package(self):
return util.create_package_from_loader(None)
@property
def files(self):
return resources.files(self.package)
def test_spec_path_joinpath(self):
self.assertIsInstance(self.files / 'a', CompatibilityFiles.OrphanPath)

View File

@ -0,0 +1,42 @@
import unittest
from importlib import resources
from . import data01
from .resources import util
class ContentsTests:
expected = {
'__init__.py',
'binary.file',
'subdirectory',
'utf-16.file',
'utf-8.file',
}
def test_contents(self):
assert self.expected <= set(resources.contents(self.data))
class ContentsDiskTests(ContentsTests, unittest.TestCase):
def setUp(self):
self.data = data01
class ContentsZipTests(ContentsTests, util.ZipSetup, unittest.TestCase):
pass
class ContentsNamespaceTests(ContentsTests, unittest.TestCase):
expected = {
# no __init__ because of namespace design
# no subdirectory as incidental difference in fixture
'binary.file',
'utf-16.file',
'utf-8.file',
}
def setUp(self):
from . import namespacedata01
self.data = namespacedata01

View File

@ -4,7 +4,7 @@ import unittest
from importlib import resources from importlib import resources
from importlib.abc import Traversable from importlib.abc import Traversable
from . import data01 from . import data01
from . import util from .resources import util
class FilesTests: class FilesTests:
@ -35,5 +35,12 @@ class OpenZipTests(FilesTests, util.ZipSetup, unittest.TestCase):
pass pass
class OpenNamespaceTests(FilesTests, unittest.TestCase):
def setUp(self):
from . import namespacedata01
self.data = namespacedata01
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

View File

@ -2,16 +2,16 @@ import unittest
from importlib import resources from importlib import resources
from . import data01 from . import data01
from . import util from .resources import util
class CommonBinaryTests(util.CommonResourceTests, unittest.TestCase): class CommonBinaryTests(util.CommonTests, unittest.TestCase):
def execute(self, package, path): def execute(self, package, path):
with resources.open_binary(package, path): with resources.open_binary(package, path):
pass pass
class CommonTextTests(util.CommonResourceTests, unittest.TestCase): class CommonTextTests(util.CommonTests, unittest.TestCase):
def execute(self, package, path): def execute(self, package, path):
with resources.open_text(package, path): with resources.open_text(package, path):
pass pass

View File

@ -3,10 +3,10 @@ import unittest
from importlib import resources from importlib import resources
from . import data01 from . import data01
from . import util from .resources import util
class CommonTests(util.CommonResourceTests, unittest.TestCase): class CommonTests(util.CommonTests, unittest.TestCase):
def execute(self, package, path): def execute(self, package, path):
with resources.path(package, path): with resources.path(package, path):
pass pass

View File

@ -2,15 +2,15 @@ import unittest
from importlib import import_module, resources from importlib import import_module, resources
from . import data01 from . import data01
from . import util from .resources import util
class CommonBinaryTests(util.CommonResourceTests, unittest.TestCase): class CommonBinaryTests(util.CommonTests, unittest.TestCase):
def execute(self, package, path): def execute(self, package, path):
resources.read_binary(package, path) resources.read_binary(package, path)
class CommonTextTests(util.CommonResourceTests, unittest.TestCase): class CommonTextTests(util.CommonTests, unittest.TestCase):
def execute(self, package, path): def execute(self, package, path):
resources.read_text(package, path) resources.read_text(package, path)
@ -55,5 +55,12 @@ class ReadZipTests(ReadTests, util.ZipSetup, unittest.TestCase):
self.assertEqual(result, b'\0\1\2\3') self.assertEqual(result, b'\0\1\2\3')
class ReadNamespaceTests(ReadTests, unittest.TestCase):
def setUp(self):
from . import namespacedata01
self.data = namespacedata01
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

View File

@ -5,7 +5,7 @@ import pathlib
from . import data01 from . import data01
from . import zipdata01, zipdata02 from . import zipdata01, zipdata02
from . import util from .resources import util
from importlib import resources, import_module from importlib import resources, import_module
from test.support import import_helper from test.support import import_helper
from test.support.os_helper import unlink from test.support.os_helper import unlink
@ -33,14 +33,14 @@ class ResourceTests:
# are not germane to this test, so just filter them out. # are not germane to this test, so just filter them out.
contents.discard('__pycache__') contents.discard('__pycache__')
self.assertEqual( self.assertEqual(
contents, sorted(contents),
{ [
'__init__.py', '__init__.py',
'subdirectory',
'utf-8.file',
'binary.file', 'binary.file',
'subdirectory',
'utf-16.file', 'utf-16.file',
}, 'utf-8.file',
],
) )

View File

@ -1,17 +1,11 @@
import abc
import builtins import builtins
import contextlib import contextlib
import errno import errno
import functools import functools
import importlib
from importlib import machinery, util, invalidate_caches from importlib import machinery, util, invalidate_caches
from importlib.abc import ResourceReader
import io
import marshal import marshal
import os import os
import os.path import os.path
from pathlib import Path, PurePath
from test import support
from test.support import import_helper from test.support import import_helper
from test.support import os_helper from test.support import os_helper
import unittest import unittest
@ -19,9 +13,6 @@ import sys
import tempfile import tempfile
import types import types
from . import data01
from . import zipdata01
BUILTINS = types.SimpleNamespace() BUILTINS = types.SimpleNamespace()
BUILTINS.good_name = None BUILTINS.good_name = None
@ -417,166 +408,3 @@ class CASEOKTestBase:
if any(x in self.importlib._bootstrap_external._os.environ if any(x in self.importlib._bootstrap_external._os.environ
for x in possibilities) != should_exist: for x in possibilities) != should_exist:
self.skipTest('os.environ changes not reflected in _os.environ') self.skipTest('os.environ changes not reflected in _os.environ')
def create_package(file, path, is_package=True, contents=()):
class Reader(ResourceReader):
def get_resource_reader(self, package):
return self
def open_resource(self, path):
self._path = path
if isinstance(file, Exception):
raise file
else:
return file
def resource_path(self, path_):
self._path = path_
if isinstance(path, Exception):
raise path
else:
return path
def is_resource(self, path_):
self._path = path_
if isinstance(path, Exception):
raise path
for entry in contents:
parts = entry.split('/')
if len(parts) == 1 and parts[0] == path_:
return True
return False
def contents(self):
if isinstance(path, Exception):
raise path
# There's no yield from in baseball, er, Python 2.
for entry in contents:
yield entry
name = 'testingpackage'
# Unfortunately importlib.util.module_from_spec() was not introduced until
# Python 3.5.
module = types.ModuleType(name)
loader = Reader()
spec = machinery.ModuleSpec(
name, loader,
origin='does-not-exist',
is_package=is_package)
module.__spec__ = spec
module.__loader__ = loader
return module
class CommonResourceTests(abc.ABC):
@abc.abstractmethod
def execute(self, package, path):
raise NotImplementedError
def test_package_name(self):
# Passing in the package name should succeed.
self.execute(data01.__name__, 'utf-8.file')
def test_package_object(self):
# Passing in the package itself should succeed.
self.execute(data01, 'utf-8.file')
def test_string_path(self):
# Passing in a string for the path should succeed.
path = 'utf-8.file'
self.execute(data01, path)
@unittest.skipIf(sys.version_info < (3, 6), 'requires os.PathLike support')
def test_pathlib_path(self):
# Passing in a pathlib.PurePath object for the path should succeed.
path = PurePath('utf-8.file')
self.execute(data01, path)
def test_absolute_path(self):
# An absolute path is a ValueError.
path = Path(__file__)
full_path = path.parent/'utf-8.file'
with self.assertRaises(ValueError):
self.execute(data01, full_path)
def test_relative_path(self):
# A relative path is a ValueError.
with self.assertRaises(ValueError):
self.execute(data01, '../data01/utf-8.file')
def test_importing_module_as_side_effect(self):
# The anchor package can already be imported.
del sys.modules[data01.__name__]
self.execute(data01.__name__, 'utf-8.file')
def test_non_package_by_name(self):
# The anchor package cannot be a module.
with self.assertRaises(TypeError):
self.execute(__name__, 'utf-8.file')
def test_non_package_by_package(self):
# The anchor package cannot be a module.
with self.assertRaises(TypeError):
module = sys.modules['test.test_importlib.util']
self.execute(module, 'utf-8.file')
@unittest.skipIf(sys.version_info < (3,), 'No ResourceReader in Python 2')
def test_resource_opener(self):
bytes_data = io.BytesIO(b'Hello, world!')
package = create_package(file=bytes_data, path=FileNotFoundError())
self.execute(package, 'utf-8.file')
self.assertEqual(package.__loader__._path, 'utf-8.file')
@unittest.skipIf(sys.version_info < (3,), 'No ResourceReader in Python 2')
def test_resource_path(self):
bytes_data = io.BytesIO(b'Hello, world!')
path = __file__
package = create_package(file=bytes_data, path=path)
self.execute(package, 'utf-8.file')
self.assertEqual(package.__loader__._path, 'utf-8.file')
def test_useless_loader(self):
package = create_package(file=FileNotFoundError(),
path=FileNotFoundError())
with self.assertRaises(FileNotFoundError):
self.execute(package, 'utf-8.file')
class ZipSetupBase:
ZIP_MODULE = None
@classmethod
def setUpClass(cls):
data_path = Path(cls.ZIP_MODULE.__file__)
data_dir = data_path.parent
cls._zip_path = str(data_dir / 'ziptestdata.zip')
sys.path.append(cls._zip_path)
cls.data = importlib.import_module('ziptestdata')
@classmethod
def tearDownClass(cls):
try:
sys.path.remove(cls._zip_path)
except ValueError:
pass
try:
del sys.path_importer_cache[cls._zip_path]
del sys.modules[cls.data.__name__]
except KeyError:
pass
try:
del cls.data
del cls._zip_path
except AttributeError:
pass
def setUp(self):
modules = import_helper.modules_setup()
self.addCleanup(import_helper.modules_cleanup, *modules)
class ZipSetup(ZipSetupBase):
ZIP_MODULE = zipdata01 # type: ignore

View File

@ -0,0 +1,5 @@
Added ``importlib.simple`` module implementing adapters from a low-level
resources reader interface to a ``TraversableResources`` interface. Legacy
API (``path``, ``contents``, ...) is now supported entirely by the
``.files()`` API with a compatibility shim supplied for resource loaders
without that functionality. Feature parity with ``importlib_resources`` 5.2.