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:
-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
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):
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()
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')
# 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')
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.
... "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
@create.register
def _(content: str, path):
- path.write_text(content)
+ path.write_text(content, encoding='utf-8')
# end from jaraco.path
--- /dev/null
+a resource
\ No newline at end of file
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):
--- /dev/null
+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
_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']
'__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',
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
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')
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)
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):
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))
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()
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:
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):
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')
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)),
+import contextlib
import sys
import unittest
import uuid
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
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']
)
)
+@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):
"""
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')
--- /dev/null
+Apply changes from `importlib_resources 5.12
+<https://importlib-resources.readthedocs.io/en/latest/history.html#v5-12-0>`_,
+including fix for ``MultiplexedPath`` to support directories in multiple
+namespaces (python/importlib_resources#265).
projects / thirdparty / Python / cpython.git / commitdiff
