Spaces:
Sleeping
Sleeping
File size: 6,065 Bytes
7d134e4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 |
import abc
import importlib
import io
import sys
import types
import pathlib
import contextlib
from ..abc import ResourceReader
from .compat.py39 import import_helper, os_helper
from . import zip as zip_
from . import _path
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 CommonTestsBase(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(self.data.__name__, 'utf-8.file')
def test_package_object(self):
"""
Passing in the package itself should succeed.
"""
self.execute(self.data, 'utf-8.file')
def test_string_path(self):
"""
Passing in a string for the path should succeed.
"""
path = 'utf-8.file'
self.execute(self.data, path)
def test_pathlib_path(self):
"""
Passing in a pathlib.PurePath object for the path should succeed.
"""
path = pathlib.PurePath('utf-8.file')
self.execute(self.data, path)
def test_importing_module_as_side_effect(self):
"""
The anchor package can already be imported.
"""
del sys.modules[self.data.__name__]
self.execute(self.data.__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.
"""
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')
fixtures = dict(
data01={
'__init__.py': '',
'binary.file': bytes(range(4)),
'utf-16.file': 'Hello, UTF-16 world!\n'.encode('utf-16'),
'utf-8.file': 'Hello, UTF-8 world!\n'.encode('utf-8'),
'subdirectory': {
'__init__.py': '',
'binary.file': bytes(range(4, 8)),
},
},
data02={
'__init__.py': '',
'one': {'__init__.py': '', 'resource1.txt': 'one resource'},
'two': {'__init__.py': '', 'resource2.txt': 'two resource'},
'subdirectory': {'subsubdir': {'resource.txt': 'a resource'}},
},
namespacedata01={
'binary.file': bytes(range(4)),
'utf-16.file': 'Hello, UTF-16 world!\n'.encode('utf-16'),
'utf-8.file': 'Hello, UTF-8 world!\n'.encode('utf-8'),
'subdirectory': {
'binary.file': bytes(range(12, 16)),
},
},
)
class ModuleSetup:
def setUp(self):
self.fixtures = contextlib.ExitStack()
self.addCleanup(self.fixtures.close)
self.fixtures.enter_context(import_helper.isolated_modules())
self.data = self.load_fixture(self.MODULE)
def load_fixture(self, module):
self.tree_on_path({module: fixtures[module]})
return importlib.import_module(module)
class ZipSetup(ModuleSetup):
MODULE = 'data01'
def tree_on_path(self, spec):
temp_dir = self.fixtures.enter_context(os_helper.temp_dir())
modules = pathlib.Path(temp_dir) / 'zipped modules.zip'
self.fixtures.enter_context(
import_helper.DirsOnSysPath(str(zip_.make_zip_file(spec, modules)))
)
class DiskSetup(ModuleSetup):
MODULE = 'data01'
def tree_on_path(self, spec):
temp_dir = self.fixtures.enter_context(os_helper.temp_dir())
_path.build(spec, pathlib.Path(temp_dir))
self.fixtures.enter_context(import_helper.DirsOnSysPath(temp_dir))
class CommonTests(DiskSetup, CommonTestsBase):
pass
|