Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion .coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ omit =
*/.tox/*
*/_itertools.py
*/_legacy.py
*/simple.py
*/_path.py
disable_warnings =
couldnt-parse
Expand Down
19 changes: 16 additions & 3 deletions importlib_resources/simple.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,14 +55,18 @@ class ResourceContainer(Traversable):
def __init__(self, reader: SimpleReader):
self.reader = reader

@property
def name(self):
return self.reader.name

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)
files = (ResourceHandle(self, name) for name in self.reader.resources())
dirs = map(ResourceContainer, self.reader.children())
return itertools.chain(files, dirs)

Expand All @@ -77,7 +81,14 @@ class ResourceHandle(Traversable):

def __init__(self, parent: ResourceContainer, name: str):
self.parent = parent
self.name = name # type: ignore[misc]
self._name = name

@property
def name(self):
return self._name

def iterdir(self):
return iter(())

def is_file(self):
return True
Expand All @@ -91,7 +102,9 @@ def open(self, mode='r', *args, **kwargs):
stream = io.TextIOWrapper(stream, *args, **kwargs)
return stream

def joinpath(self, name):
def joinpath(self, *descendants):
if not descendants:
return self
raise RuntimeError("Cannot traverse into a resource")


Expand Down
79 changes: 79 additions & 0 deletions importlib_resources/tests/test_simple.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
import io
import unittest

from importlib_resources.simple import TraversableReader


class MemoryReader(TraversableReader):
def __init__(self, package, resources=None, children=None):
self._package = package
self._resources = resources or {}
self._children = children or []

@property
def package(self):
return self._package

def children(self):
return self._children

def resources(self):
return list(self._resources)

def open_binary(self, resource):
return io.BytesIO(self._resources[resource])


class SimpleReaderTest(unittest.TestCase):
def setUp(self):
child = MemoryReader('sample.package.child', {'nested.txt': b'nested'})
self.reader = MemoryReader(
'sample.package',
{'data.txt': 'hello, café'.encode()},
[child],
)

def test_files(self):
root = self.reader.files()
assert root.name == 'package'
assert root.is_dir()
assert not root.is_file()
assert root.joinpath() is root
with self.assertRaises(IsADirectoryError):
root.open()

def test_iterdir(self):
entries = {item.name: item for item in self.reader.files().iterdir()}
assert set(entries) == {'data.txt', 'child'}
assert entries['data.txt'].is_file()
assert not entries['data.txt'].is_dir()
assert entries['child'].is_dir()
assert not entries['child'].is_file()
assert set(self.reader.contents()) == {'data.txt', 'child'}

def test_read_resource(self):
resource = self.reader.files().joinpath('data.txt')
assert resource.read_bytes() == 'hello, café'.encode()
assert resource.read_text(encoding='utf-8') == 'hello, café'
with self.reader.open_resource('data.txt') as stream:
assert stream.read() == 'hello, café'.encode()
assert self.reader.is_resource('data.txt')
assert not self.reader.is_resource('child')

def test_nested_resource(self):
root = self.reader.files()
assert root.joinpath('child', 'nested.txt').read_bytes() == b'nested'
assert root.joinpath('child/nested.txt').read_bytes() == b'nested'
assert (root / 'child' / 'nested.txt').read_bytes() == b'nested'

def test_resource_has_no_children(self):
resource = self.reader.files().joinpath('data.txt')
assert list(resource.iterdir()) == []
assert resource.joinpath() is resource
with self.assertRaisesRegex(RuntimeError, 'Cannot traverse into a resource'):
resource.joinpath('child')

def test_empty_reader(self):
root = MemoryReader('empty').files()
assert root.name == 'empty'
assert list(root.iterdir()) == []
1 change: 1 addition & 0 deletions newsfragments/+simple-reader-adapter.bugfix.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix the SimpleReader adapters to support resource enumeration, traversal, and reading through TraversableReader.