diff options
author | Jason R. Coombs <jaraco@jaraco.com> | 2023-02-18 21:29:22 (GMT) |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-02-18 21:29:22 (GMT) |
commit | 5170caf3059fdacc92d7370eecb9fe4f0c5a1c76 (patch) | |
tree | b294f3f9a122e4ffc520ec4ef22aabadc34144a3 /Lib | |
parent | 128379b8cdb88a6d3d7fed24df082c9a654b3fb8 (diff) | |
download | cpython-5170caf3059fdacc92d7370eecb9fe4f0c5a1c76.zip cpython-5170caf3059fdacc92d7370eecb9fe4f0c5a1c76.tar.gz cpython-5170caf3059fdacc92d7370eecb9fe4f0c5a1c76.tar.bz2 |
gh-97930: Apply changes from importlib_resources 5.12. (GH-102010)
Diffstat (limited to 'Lib')
-rw-r--r-- | Lib/importlib/resources/_adapters.py | 4 | ||||
-rw-r--r-- | Lib/importlib/resources/_itertools.py | 69 | ||||
-rw-r--r-- | Lib/importlib/resources/readers.py | 36 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/_path.py | 18 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/data02/subdirectory/subsubdir/resource.txt | 1 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_compatibilty_files.py | 6 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_custom.py | 46 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_files.py | 4 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_open.py | 14 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_path.py | 15 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_read.py | 12 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_reader.py | 11 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/test_resource.py | 100 | ||||
-rw-r--r-- | Lib/test/test_importlib/resources/util.py | 28 |
14 files changed, 237 insertions, 127 deletions
diff --git a/Lib/importlib/resources/_adapters.py b/Lib/importlib/resources/_adapters.py index f22f6bc..50688fb 100644 --- a/Lib/importlib/resources/_adapters.py +++ b/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: diff --git a/Lib/importlib/resources/_itertools.py b/Lib/importlib/resources/_itertools.py index cce0558..7b775ef5a 100644 --- a/Lib/importlib/resources/_itertools.py +++ b/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 diff --git a/Lib/importlib/resources/readers.py b/Lib/importlib/resources/readers.py index 80cb320..c3cdf76 100644 --- a/Lib/importlib/resources/readers.py +++ b/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 @@ class ZipReader(abc.TraversableResources): 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 @@ class MultiplexedPath(abc.Traversable): 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 @@ class MultiplexedPath(abc.Traversable): # 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') diff --git a/Lib/test/test_importlib/resources/_path.py b/Lib/test/test_importlib/resources/_path.py index c630e4d..1f97c96 100644 --- a/Lib/test/test_importlib/resources/_path.py +++ b/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 diff --git a/Lib/test/test_importlib/resources/data02/subdirectory/subsubdir/resource.txt b/Lib/test/test_importlib/resources/data02/subdirectory/subsubdir/resource.txt new file mode 100644 index 0000000..48f587a --- /dev/null +++ b/Lib/test/test_importlib/resources/data02/subdirectory/subsubdir/resource.txt @@ -0,0 +1 @@ +a resource
\ No newline at end of file diff --git a/Lib/test/test_importlib/resources/test_compatibilty_files.py b/Lib/test/test_importlib/resources/test_compatibilty_files.py index 6fa18a2..bcf608d 100644 --- a/Lib/test/test_importlib/resources/test_compatibilty_files.py +++ b/Lib/test/test_importlib/resources/test_compatibilty_files.py @@ -64,11 +64,13 @@ class CompatibilityFilesTests(unittest.TestCase): 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): diff --git a/Lib/test/test_importlib/resources/test_custom.py b/Lib/test/test_importlib/resources/test_custom.py new file mode 100644 index 0000000..7312720 --- /dev/null +++ b/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 diff --git a/Lib/test/test_importlib/resources/test_files.py b/Lib/test/test_importlib/resources/test_files.py index fe813ae..1450cfb 100644 --- a/Lib/test/test_importlib/resources/test_files.py +++ b/Lib/test/test_importlib/resources/test_files.py @@ -85,7 +85,7 @@ class ModulesFilesTests(SiteDir, unittest.TestCase): _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 @@ class ImplicitContextFilesTests(SiteDir, unittest.TestCase): '__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', diff --git a/Lib/test/test_importlib/resources/test_open.py b/Lib/test/test_importlib/resources/test_open.py index 0554c41..86becb4 100644 --- a/Lib/test/test_importlib/resources/test_open.py +++ b/Lib/test/test_importlib/resources/test_open.py @@ -15,7 +15,7 @@ class CommonBinaryTests(util.CommonTests, unittest.TestCase): 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 @@ class OpenTests: 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 @@ class OpenTests: 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 @@ class OpenTests: 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): diff --git a/Lib/test/test_importlib/resources/test_path.py b/Lib/test/test_importlib/resources/test_path.py index adcf75f..34a6bdd 100644 --- a/Lib/test/test_importlib/resources/test_path.py +++ b/Lib/test/test_importlib/resources/test_path.py @@ -14,9 +14,12 @@ class CommonTests(util.CommonTests, unittest.TestCase): 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 @@ class PathMemoryTests(PathTests, unittest.TestCase): 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() diff --git a/Lib/test/test_importlib/resources/test_read.py b/Lib/test/test_importlib/resources/test_read.py index 0ca8ee9..0889826 100644 --- a/Lib/test/test_importlib/resources/test_read.py +++ b/Lib/test/test_importlib/resources/test_read.py @@ -12,7 +12,7 @@ class CommonBinaryTests(util.CommonTests, unittest.TestCase): class CommonTextTests(util.CommonTests, unittest.TestCase): def execute(self, package, path): - resources.files(package).joinpath(path).read_text() + resources.files(package).joinpath(path).read_text(encoding='utf-8') class ReadTests: @@ -21,7 +21,11 @@ class ReadTests: self.assertEqual(result, b'\0\1\2\3') def test_read_text_default_encoding(self): - result = resources.files(self.data).joinpath('utf-8.file').read_text() + result = ( + resources.files(self.data) + .joinpath('utf-8.file') + .read_text(encoding='utf-8') + ) self.assertEqual(result, 'Hello, UTF-8 world!\n') def test_read_text_given_encoding(self): @@ -33,7 +37,9 @@ class ReadTests: self.assertEqual(result, 'Hello, UTF-16 world!\n') def test_read_text_with_errors(self): - # Raises UnicodeError without the 'errors' argument. + """ + Raises UnicodeError without the 'errors' argument. + """ target = resources.files(self.data) / 'utf-16.file' self.assertRaises(UnicodeError, target.read_text, encoding='utf-8') result = target.read_text(encoding='utf-8', errors='ignore') diff --git a/Lib/test/test_importlib/resources/test_reader.py b/Lib/test/test_importlib/resources/test_reader.py index 4fd9e6b..8670f72 100644 --- a/Lib/test/test_importlib/resources/test_reader.py +++ b/Lib/test/test_importlib/resources/test_reader.py @@ -81,6 +81,17 @@ class MultiplexedPathTest(unittest.TestCase): path = MultiplexedPath(self.folder) assert not path.joinpath('imaginary/foo.py').exists() + def test_join_path_common_subdir(self): + prefix = os.path.abspath(os.path.join(__file__, '..')) + data01 = os.path.join(prefix, 'data01') + data02 = os.path.join(prefix, 'data02') + path = MultiplexedPath(data01, data02) + self.assertIsInstance(path.joinpath('subdirectory'), MultiplexedPath) + self.assertEqual( + str(path.joinpath('subdirectory', 'subsubdir'))[len(prefix) + 1 :], + os.path.join('data02', 'subdirectory', 'subsubdir'), + ) + def test_repr(self): self.assertEqual( repr(MultiplexedPath(self.folder)), diff --git a/Lib/test/test_importlib/resources/test_resource.py b/Lib/test/test_importlib/resources/test_resource.py index f7e3abb..6f75cf5 100644 --- a/Lib/test/test_importlib/resources/test_resource.py +++ b/Lib/test/test_importlib/resources/test_resource.py @@ -1,3 +1,4 @@ +import contextlib import sys import unittest import uuid @@ -7,7 +8,7 @@ from . import data01 from . import zipdata01, zipdata02 from . import util from importlib import resources, import_module -from test.support import import_helper +from test.support import import_helper, os_helper from test.support.os_helper import unlink @@ -69,10 +70,12 @@ class ResourceLoaderTests(unittest.TestCase): class ResourceCornerCaseTests(unittest.TestCase): def test_package_has_no_reader_fallback(self): - # Test odd ball packages which: + """ + Test odd ball packages which: # 1. Do not have a ResourceReader as a loader # 2. Are not on the file system # 3. Are not in a zip file + """ module = util.create_package( file=data01, path=data01.__file__, contents=['A', 'B', 'C'] ) @@ -138,82 +141,71 @@ class ResourceFromZipsTest02(util.ZipSetupBase, unittest.TestCase): ) +@contextlib.contextmanager +def zip_on_path(dir): + data_path = pathlib.Path(zipdata01.__file__) + source_zip_path = data_path.parent.joinpath('ziptestdata.zip') + zip_path = pathlib.Path(dir) / f'{uuid.uuid4()}.zip' + zip_path.write_bytes(source_zip_path.read_bytes()) + sys.path.append(str(zip_path)) + import_module('ziptestdata') + + try: + yield + finally: + with contextlib.suppress(ValueError): + sys.path.remove(str(zip_path)) + + with contextlib.suppress(KeyError): + del sys.path_importer_cache[str(zip_path)] + del sys.modules['ziptestdata'] + + with contextlib.suppress(OSError): + unlink(zip_path) + + class DeletingZipsTest(unittest.TestCase): """Having accessed resources in a zip file should not keep an open reference to the zip. """ - ZIP_MODULE = zipdata01 - def setUp(self): + self.fixtures = contextlib.ExitStack() + self.addCleanup(self.fixtures.close) + modules = import_helper.modules_setup() self.addCleanup(import_helper.modules_cleanup, *modules) - data_path = pathlib.Path(self.ZIP_MODULE.__file__) - data_dir = data_path.parent - self.source_zip_path = data_dir / 'ziptestdata.zip' - self.zip_path = pathlib.Path(f'{uuid.uuid4()}.zip').absolute() - self.zip_path.write_bytes(self.source_zip_path.read_bytes()) - sys.path.append(str(self.zip_path)) - self.data = import_module('ziptestdata') - - def tearDown(self): - try: - sys.path.remove(str(self.zip_path)) - except ValueError: - pass - - try: - del sys.path_importer_cache[str(self.zip_path)] - del sys.modules[self.data.__name__] - except KeyError: - pass - - try: - unlink(self.zip_path) - except OSError: - # If the test fails, this will probably fail too - pass + temp_dir = self.fixtures.enter_context(os_helper.temp_dir()) + self.fixtures.enter_context(zip_on_path(temp_dir)) def test_iterdir_does_not_keep_open(self): - c = [item.name for item in resources.files('ziptestdata').iterdir()] - self.zip_path.unlink() - del c + [item.name for item in resources.files('ziptestdata').iterdir()] def test_is_file_does_not_keep_open(self): - c = resources.files('ziptestdata').joinpath('binary.file').is_file() - self.zip_path.unlink() - del c + resources.files('ziptestdata').joinpath('binary.file').is_file() def test_is_file_failure_does_not_keep_open(self): - c = resources.files('ziptestdata').joinpath('not-present').is_file() - self.zip_path.unlink() - del c + resources.files('ziptestdata').joinpath('not-present').is_file() @unittest.skip("Desired but not supported.") def test_as_file_does_not_keep_open(self): # pragma: no cover - c = resources.as_file(resources.files('ziptestdata') / 'binary.file') - self.zip_path.unlink() - del c + resources.as_file(resources.files('ziptestdata') / 'binary.file') def test_entered_path_does_not_keep_open(self): - # This is what certifi does on import to make its bundle - # available for the process duration. - c = resources.as_file( - resources.files('ziptestdata') / 'binary.file' - ).__enter__() - self.zip_path.unlink() - del c + """ + Mimic what certifi does on import to make its bundle + available for the process duration. + """ + resources.as_file(resources.files('ziptestdata') / 'binary.file').__enter__() def test_read_binary_does_not_keep_open(self): - c = resources.files('ziptestdata').joinpath('binary.file').read_bytes() - self.zip_path.unlink() - del c + resources.files('ziptestdata').joinpath('binary.file').read_bytes() def test_read_text_does_not_keep_open(self): - c = resources.files('ziptestdata').joinpath('utf-8.file').read_text() - self.zip_path.unlink() - del c + resources.files('ziptestdata').joinpath('utf-8.file').read_text( + encoding='utf-8' + ) class ResourceFromNamespaceTest01(unittest.TestCase): diff --git a/Lib/test/test_importlib/resources/util.py b/Lib/test/test_importlib/resources/util.py index 1e72b91..dbe6ee8 100644 --- a/Lib/test/test_importlib/resources/util.py +++ b/Lib/test/test_importlib/resources/util.py @@ -80,32 +80,44 @@ class CommonTests(metaclass=abc.ABCMeta): """ def test_package_name(self): - # Passing in the package name should succeed. + """ + 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. + """ + 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. + """ + 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. + """ + Passing in a pathlib.PurePath object for the path should succeed. + """ path = pathlib.PurePath('utf-8.file') self.execute(data01, path) def test_importing_module_as_side_effect(self): - # The anchor package can already be imported. + """ + The anchor package can already be imported. + """ del sys.modules[data01.__name__] self.execute(data01.__name__, '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. + """ + 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') |