518483918

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/PyFilesystem/pyfilesystem2/commit/<a class=hub.com/PyFilesystem/pyfilesystem2/commit/<a class="double-link" href="https://git"><a class=hub.com/PyFilesystem/pyfilesystem2/commit/<a class="double-link" href="https://git"><a class=hub.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fdab98e5cf7">0ebc915be<a href="https://github.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fdab98e5cf7">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/PyFilesystem/pyfilesystem2/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/PyFilesystem/pyfilesystem2/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fdab98e5cf7&quot;&gt;0ebc915be&lt;/a&gt;&lt;a href=&quot;https://github.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fdab98e5cf7&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fda&lt;/a&gt;b98e5cf7&quot;>&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/PyFilesystem/pyfilesystem2/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/PyFilesystem/pyfilesystem2/commit/<a class="double-link" href="https://github.com/PyFilesystem/pyfilesystem2/commit/3624d47c5a8dc2149fe542f9203c391184154d2a">3624d47c5</a><a href="https://github.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fdab98e5cf7">&amp;quot;&amp;gt;3624d47c5&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;3624d47c5&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/PyFilesystem/pyfilesystem2/commit/0ebc915bed11b74341ba93a5656c3fdab98e5&lt;/a&gt;cf7&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/PyFilesystem/pyfilesystem2/commit/<a class="double-link" href="https://github.com/PyFilesystem/pyfilesystem2/commit/89ef78f54">89ef78f54">89ef78f54</a>

Pull Request Pull Request #448: Replace Travis-CI with GitHub Actions

Run Details

1304 of 1435 branches covered (90.87%)

51 of 60 new or added lines in 12 files covered. (85.0%)

4819 of 5053 relevant lines covered (95.37%)

7.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.7

/fs/memoryfs.py

"""Manage a volatile in-memory filesystem.
"""
from __future__ import absolute_import
from __future__ import unicode_literals

import contextlib
import io
import os
import time
import typing
from collections import OrderedDict
from threading import RLock

import six

from . import errors
from .base import FS
from .enums import ResourceType, Seek
from .info import Info
from .mode import Mode
from .path import iteratepath
from .path import normpath
from .path import split
from ._typing import overload

if typing.TYPE_CHECKING:
    import array
    import mmap
    from typing import (
        Any,
        BinaryIO,
        Collection,
        Dict,
        Iterable,
        Iterator,
        List,
        Optional,
        SupportsInt,
        Union,
        Text,
    )
    from .base import _OpendirFactory
    from .info import RawInfo
    from .permissions import Permissions
    from .subfs import SubFS

    _M = typing.TypeVar("_M", bound="MemoryFS")


@six.python_2_unicode_compatible
class _MemoryFile(io.RawIOBase):
    def __init__(self, path, memory_fs, mode, dir_entry):
        # type: (Text, MemoryFS, Text, _DirEntry) -> None
        super(_MemoryFile, self).__init__()
        self._path = path
        self._memory_fs = memory_fs
        self._mode = Mode(mode)
        self._dir_entry = dir_entry

        # We are opening a file - dir_entry.bytes_file is not None
        self._bytes_io = typing.cast(io.BytesIO, dir_entry.bytes_file)

        self.accessed_time = time.time()
        self.modified_time = time.time()
        self.pos = 0

        if self._mode.truncate:
            with self._dir_entry.lock:
                self._bytes_io.seek(0)
                self._bytes_io.truncate()
        elif self._mode.appending:
            with self._dir_entry.lock:
                self._bytes_io.seek(0, os.SEEK_END)
                self.pos = self._bytes_io.tell()

    def __str__(self):
        # type: () -> str
        _template = "<memoryfile '{path}' '{mode}'>"
        return _template.format(path=self._path, mode=self._mode)

    @property
    def mode(self):
        # type: () -> Text
        return self._mode.to_platform_bin()

    @contextlib.contextmanager
    def _seek_lock(self):
        # type: () -> Iterator[None]
        with self._dir_entry.lock:
            self._bytes_io.seek(self.pos)
            yield
            self.pos = self._bytes_io.tell()

    def on_modify(self):  # noqa: D401
        # type: () -> None
        """Called when file data is modified."""
        self._dir_entry.modified_time = self.modified_time = time.time()

    def on_access(self):  # noqa: D401
        # type: () -> None
        """Called when file is accessed."""
        self._dir_entry.accessed_time = self.accessed_time = time.time()

    def flush(self):
        # type: () -> None
        pass

    def __iter__(self):
        # type: () -> typing.Iterator[bytes]
        self._bytes_io.seek(self.pos)
        for line in self._bytes_io:
            yield line

    def next(self):
        # type: () -> bytes
        with self._seek_lock():
            self.on_access()
            return next(self._bytes_io)

    __next__ = next

    def readline(self, size=None):
        # type: (Optional[int]) -> bytes
        if not self._mode.reading:
            raise IOError("File not open for reading")
        with self._seek_lock():
            self.on_access()
            return self._bytes_io.readline(size)

    def close(self):
        # type: () -> None
        if not self.closed:
            with self._dir_entry.lock:
                self._dir_entry.remove_open_file(self)
                super(_MemoryFile, self).close()

    def read(self, size=None):
        # type: (Optional[int]) -> bytes
        if not self._mode.reading:
            raise IOError("File not open for reading")
        with self._seek_lock():
            self.on_access()
            return self._bytes_io.read(size)

    def readable(self):
        # type: () -> bool
        return self._mode.reading

    def readinto(self, buffer):
        # type (bytearray) -> Optional[int]
        if not self._mode.reading:
            raise IOError("File not open for reading")
        with self._seek_lock():
            self.on_access()
            return self._bytes_io.readinto(buffer)

    def readlines(self, hint=-1):
        # type: (int) -> List[bytes]
        if not self._mode.reading:
            raise IOError("File not open for reading")
        with self._seek_lock():
            self.on_access()
            return self._bytes_io.readlines(hint)

    def seekable(self):
        # type: () -> bool
        return True

    def seek(self, pos, whence=Seek.set):
        # type: (int, SupportsInt) -> int
        # NOTE(@althonos): allows passing both Seek.set and os.SEEK_SET
        with self._seek_lock():
            self.on_access()
            return self._bytes_io.seek(pos, int(whence))

    def tell(self):
        # type: () -> int
        return self.pos

    def truncate(self, size=None):
        # type: (Optional[int]) -> int
        with self._seek_lock():
            self.on_modify()
            new_size = self._bytes_io.truncate(size)
            if size is not None and self._bytes_io.tell() < size:
                file_size = self._bytes_io.seek(0, os.SEEK_END)
                self._bytes_io.write(b"\0" * (size - file_size))
                self._bytes_io.seek(-size + file_size, os.SEEK_END)
            return size or new_size

    def writable(self):
        # type: () -> bool
        return self._mode.writing

    def write(self, data):
        # type: (Union[bytes, memoryview, array.array[Any], mmap.mmap]) -> int
        if not self._mode.writing:
            raise IOError("File not open for writing")
        with self._seek_lock():
            self.on_modify()
            return self._bytes_io.write(data)

    def writelines(self, sequence):
        # type: (Iterable[Union[bytes, memoryview, array.array[Any], mmap.mmap]]) -> None  # noqa: E501
        with self._seek_lock():
            self.on_modify()
            self._bytes_io.writelines(sequence)


class _DirEntry(object):
    def __init__(self, resource_type, name):
        # type: (ResourceType, Text) -> None
        self.resource_type = resource_type
        self.name = name
        self._dir = OrderedDict()  # type: typing.MutableMapping[Text, _DirEntry]
        self._open_files = []  # type: typing.MutableSequence[_MemoryFile]
        self._bytes_file = None  # type: Optional[io.BytesIO]
        self.lock = RLock()

        current_time = time.time()
        self.created_time = current_time
        self.accessed_time = current_time
        self.modified_time = current_time

        if not self.is_dir:
            self._bytes_file = io.BytesIO()

    @property
    def bytes_file(self):
        # type: () -> Optional[io.BytesIO]
        return self._bytes_file

    @property
    def is_dir(self):
        # type: () -> bool
        return self.resource_type == ResourceType.directory

    @property
    def size(self):
        # type: () -> int
        with self.lock:
            if self.is_dir:
                return 0
            else:
                _bytes_file = typing.cast(io.BytesIO, self._bytes_file)
                _bytes_file.seek(0, os.SEEK_END)
                return _bytes_file.tell()

    @overload
    def get_entry(self, name, default):  # noqa: F811
        # type: (Text, _DirEntry) -> _DirEntry
        pass

    @overload
    def get_entry(self, name):  # noqa: F811
        # type: (Text) -> Optional[_DirEntry]
        pass

    @overload
    def get_entry(self, name, default):  # noqa: F811
        # type: (Text, None) -> Optional[_DirEntry]
        pass

    def get_entry(self, name, default=None):  # noqa: F811
        # type: (Text, Optional[_DirEntry]) -> Optional[_DirEntry]
        assert self.is_dir, "must be a directory"
        return self._dir.get(name, default)

    def set_entry(self, name, dir_entry):
        # type: (Text, _DirEntry) -> None
        self._dir[name] = dir_entry

    def remove_entry(self, name):
        # type: (Text) -> None
        del self._dir[name]

    def __contains__(self, name):
        # type: (object) -> bool
        return name in self._dir

    def __len__(self):
        # type: () -> int
        return len(self._dir)

    def list(self):
        # type: () -> List[Text]
        return list(self._dir.keys())

    def add_open_file(self, memory_file):
        # type: (_MemoryFile) -> None
        self._open_files.append(memory_file)

    def remove_open_file(self, memory_file):
        # type: (_MemoryFile) -> None
        self._open_files.remove(memory_file)


@six.python_2_unicode_compatible
class MemoryFS(FS):
    """A filesystem that stored in memory.

    Memory filesystems are useful for caches, temporary data stores,
    unit testing, etc. Since all the data is in memory, they are very
    fast, but non-permanent. The `MemoryFS` constructor takes no
    arguments.

    Examples:
        Create with the constructor::

            >>> from fs.memoryfs import MemoryFS
            >>> mem_fs = MemoryFS()

        Or via an FS URL::

            >>> import fs
            >>> mem_fs = fs.open_fs('mem://')

    """

    _meta = {
        "case_insensitive": False,
        "invalid_path_chars": "\0",
        "network": False,
        "read_only": False,
        "thread_safe": True,
        "unicode_paths": True,
        "virtual": False,
    }  # type: Dict[Text, Union[Text, int, bool, None]]

    def __init__(self):
        # type: () -> None
        """Create an in-memory filesystem."""
        self._meta = self._meta.copy()
        self.root = self._make_dir_entry(ResourceType.directory, "")
        super(MemoryFS, self).__init__()

    def __repr__(self):
        # type: () -> str
        return "MemoryFS()"

    def __str__(self):
        # type: () -> str
        return "<memfs>"

    def _make_dir_entry(self, resource_type, name):
        # type: (ResourceType, Text) -> _DirEntry
        return _DirEntry(resource_type, name)

    def _get_dir_entry(self, dir_path):
        # type: (Text) -> Optional[_DirEntry]
        """Get a directory entry, or `None` if one doesn't exist."""
        with self._lock:
            dir_path = normpath(dir_path)
            current_entry = self.root  # type: Optional[_DirEntry]
            for path_component in iteratepath(dir_path):
                if current_entry is None:
                    return None
                if not current_entry.is_dir:
                    return None
                current_entry = current_entry.get_entry(path_component)
            return current_entry

    def close(self):
        # type: () -> None
        if not self._closed:
            del self.root
        return super(MemoryFS, self).close()

    def getinfo(self, path, namespaces=None):
        # type: (Text, Optional[Collection[Text]]) -> Info
        namespaces = namespaces or ()
        _path = self.validatepath(path)
        dir_entry = self._get_dir_entry(_path)
        if dir_entry is None:
            raise errors.ResourceNotFound(path)
        info = {"basic": {"name": dir_entry.name, "is_dir": dir_entry.is_dir}}
        if "details" in namespaces:
            info["details"] = {
                "_write": ["accessed", "modified"],
                "type": int(dir_entry.resource_type),
                "size": dir_entry.size,
                "accessed": dir_entry.accessed_time,
                "modified": dir_entry.modified_time,
                "created": dir_entry.created_time,
            }
        return Info(info)

    def listdir(self, path):
        # type: (Text) -> List[Text]
        self.check()
        _path = self.validatepath(path)
        with self._lock:
            dir_entry = self._get_dir_entry(_path)
            if dir_entry is None:
                raise errors.ResourceNotFound(path)
            if not dir_entry.is_dir:
                raise errors.DirectoryExpected(path)
            return dir_entry.list()

    if typing.TYPE_CHECKING:

        def opendir(self, path, factory=None):
            # type: (_M, Text, Optional[_OpendirFactory]) -> SubFS[_M]
            pass

    def makedir(
        self,  # type: _M
        path,  # type: Text
        permissions=None,  # type: Optional[Permissions]
        recreate=False,  # type: bool
    ):
        # type: (...) -> SubFS[_M]
        _path = self.validatepath(path)
        with self._lock:
            if _path == "/":
                if recreate:
                    return self.opendir(path)
                else:
                    raise errors.DirectoryExists(path)

            dir_path, dir_name = split(_path)

            parent_dir = self._get_dir_entry(dir_path)
            if parent_dir is None:
                raise errors.ResourceNotFound(path)

            dir_entry = parent_dir.get_entry(dir_name)
            if dir_entry is not None and not recreate:
                raise errors.DirectoryExists(path)

            if dir_entry is None:
                new_dir = self._make_dir_entry(ResourceType.directory, dir_name)
                parent_dir.set_entry(dir_name, new_dir)
            return self.opendir(path)

    def openbin(self, path, mode="r", buffering=-1, **options):
        # type: (Text, Text, int, **Any) -> BinaryIO
        _mode = Mode(mode)
        _mode.validate_bin()
        _path = self.validatepath(path)
        dir_path, file_name = split(_path)

        if not file_name:
            raise errors.FileExpected(path)

        with self._lock:
            parent_dir_entry = self._get_dir_entry(dir_path)
            if parent_dir_entry is None or not parent_dir_entry.is_dir:
                raise errors.ResourceNotFound(path)

            if _mode.create:
                if file_name not in parent_dir_entry:
                    file_dir_entry = self._make_dir_entry(ResourceType.file, file_name)
                    parent_dir_entry.set_entry(file_name, file_dir_entry)
                else:
                    file_dir_entry = self._get_dir_entry(_path)  # type: ignore
                    if _mode.exclusive:
                        raise errors.FileExists(path)

                if file_dir_entry.is_dir:
                    raise errors.FileExpected(path)

                mem_file = _MemoryFile(
                    path=_path, memory_fs=self, mode=mode, dir_entry=file_dir_entry
                )

                file_dir_entry.add_open_file(mem_file)
                return mem_file  # type: ignore

            if file_name not in parent_dir_entry:
                raise errors.ResourceNotFound(path)

            file_dir_entry = parent_dir_entry.get_entry(file_name)  # type: ignore
            if file_dir_entry.is_dir:
                raise errors.FileExpected(path)

            mem_file = _MemoryFile(
                path=_path, memory_fs=self, mode=mode, dir_entry=file_dir_entry
            )
            file_dir_entry.add_open_file(mem_file)
            return mem_file  # type: ignore

    def remove(self, path):
        # type: (Text) -> None
        _path = self.validatepath(path)

        with self._lock:
            dir_path, file_name = split(_path)
            parent_dir_entry = self._get_dir_entry(dir_path)

            if parent_dir_entry is None or file_name not in parent_dir_entry:
                raise errors.ResourceNotFound(path)

            file_dir_entry = typing.cast(_DirEntry, self._get_dir_entry(_path))
            if file_dir_entry.is_dir:
                raise errors.FileExpected(path)

            parent_dir_entry.remove_entry(file_name)

    def removedir(self, path):
        # type: (Text) -> None
        _path = self.validatepath(path)

        if _path == "/":
            raise errors.RemoveRootError()

        with self._lock:
            dir_path, file_name = split(_path)
            parent_dir_entry = self._get_dir_entry(dir_path)

            if parent_dir_entry is None or file_name not in parent_dir_entry:
                raise errors.ResourceNotFound(path)

            dir_dir_entry = typing.cast(_DirEntry, self._get_dir_entry(_path))
            if not dir_dir_entry.is_dir:
                raise errors.DirectoryExpected(path)

            if len(dir_dir_entry):
                raise errors.DirectoryNotEmpty(path)

            parent_dir_entry.remove_entry(file_name)

    def setinfo(self, path, info):
        # type: (Text, RawInfo) -> None
        _path = self.validatepath(path)
        with self._lock:
            dir_path, file_name = split(_path)
            parent_dir_entry = self._get_dir_entry(dir_path)

            if parent_dir_entry is None or file_name not in parent_dir_entry:
                raise errors.ResourceNotFound(path)

            resource_entry = typing.cast(
                _DirEntry, parent_dir_entry.get_entry(file_name)
            )

            if "details" in info:
                details = info["details"]
                if "accessed" in details:
                    resource_entry.accessed_time = details["accessed"]  # type: ignore
                if "modified" in details:
                    resource_entry.modified_time = details["modified"]  # type: ignore

1	"""Manage a volatile in-memory filesystem.
2	"""
3	from __future__ import absolute_import	9✔
4	from __future__ import unicode_literals	9✔
5
6	import contextlib	9✔
7	import io	9✔
8	import os	9✔
9	import time	9✔
10	import typing	9✔
11	from collections import OrderedDict	9✔
12	from threading import RLock	9✔
13
14	import six	9✔
15
16	from . import errors	9✔
17	from .base import FS	9✔
18	from .enums import ResourceType, Seek	9✔
19	from .info import Info	9✔
20	from .mode import Mode	9✔
21	from .path import iteratepath	9✔
22	from .path import normpath	9✔
23	from .path import split	9✔
24	from ._typing import overload	9✔
25
26	if typing.TYPE_CHECKING:	9!
NEW 27	import array	×
NEW 28	import mmap	×
29	from typing import (	×
30	Any,
31	BinaryIO,
32	Collection,
33	Dict,
34	Iterable,
35	Iterator,
36	List,
37	Optional,
38	SupportsInt,
39	Union,
40	Text,
41	)
42	from .base import _OpendirFactory	×
43	from .info import RawInfo	×
44	from .permissions import Permissions	×
45	from .subfs import SubFS	×
46
47	_M = typing.TypeVar("_M", bound="MemoryFS")	×
48
49
50	@six.python_2_unicode_compatible	9✔
51	class _MemoryFile(io.RawIOBase):	9✔
52	def __init__(self, path, memory_fs, mode, dir_entry):	9✔
53	# type: (Text, MemoryFS, Text, _DirEntry) -> None
54	super(_MemoryFile, self).__init__()	9✔
55	self._path = path	9✔
56	self._memory_fs = memory_fs	9✔
57	self._mode = Mode(mode)	9✔
58	self._dir_entry = dir_entry	9✔
59
60	# We are opening a file - dir_entry.bytes_file is not None
61	self._bytes_io = typing.cast(io.BytesIO, dir_entry.bytes_file)	9✔
62
63	self.accessed_time = time.time()	9✔
64	self.modified_time = time.time()	9✔
65	self.pos = 0	9✔
66
67	if self._mode.truncate:	9✔
68	with self._dir_entry.lock:	9✔
69	self._bytes_io.seek(0)	9✔
70	self._bytes_io.truncate()	9✔
71	elif self._mode.appending:	9✔
72	with self._dir_entry.lock:	9✔
73	self._bytes_io.seek(0, os.SEEK_END)	9✔
74	self.pos = self._bytes_io.tell()	9✔
75
76	def __str__(self):	9✔
77	# type: () -> str
78	_template = "<memoryfile '{path}' '{mode}'>"	9✔
79	return _template.format(path=self._path, mode=self._mode)	9✔
80
81	@property	9✔
82	def mode(self):	2✔
83	# type: () -> Text
84	return self._mode.to_platform_bin()	9✔
85
86	@contextlib.contextmanager	9✔
87	def _seek_lock(self):	2✔
88	# type: () -> Iterator[None]
89	with self._dir_entry.lock:	9✔
90	self._bytes_io.seek(self.pos)	9✔
91	yield	9✔
92	self.pos = self._bytes_io.tell()	9✔
93
94	def on_modify(self): # noqa: D401	9✔
95	# type: () -> None
96	"""Called when file data is modified."""
97	self._dir_entry.modified_time = self.modified_time = time.time()	9✔
98
99	def on_access(self): # noqa: D401	9✔
100	# type: () -> None
101	"""Called when file is accessed."""
102	self._dir_entry.accessed_time = self.accessed_time = time.time()	9✔
103
104	def flush(self):	9✔
105	# type: () -> None
106	pass	9✔
107
108	def __iter__(self):	9✔
109	# type: () -> typing.Iterator[bytes]
110	self._bytes_io.seek(self.pos)	9✔
111	for line in self._bytes_io:	9✔
112	yield line	9✔
113
114	def next(self):	9✔
115	# type: () -> bytes
116	with self._seek_lock():	9✔
117	self.on_access()	9✔
118	return next(self._bytes_io)	9✔
119
120	__next__ = next	9✔
121
122	def readline(self, size=None):	9✔
123	# type: (Optional[int]) -> bytes
124	if not self._mode.reading:	9!
125	raise IOError("File not open for reading")	×
126	with self._seek_lock():	9✔
127	self.on_access()	9✔
128	return self._bytes_io.readline(size)	9✔
129
130	def close(self):	9✔
131	# type: () -> None
132	if not self.closed:	9!
133	with self._dir_entry.lock:	9✔
134	self._dir_entry.remove_open_file(self)	9✔
135	super(_MemoryFile, self).close()	9✔
136
137	def read(self, size=None):	9✔
138	# type: (Optional[int]) -> bytes
139	if not self._mode.reading:	9✔
140	raise IOError("File not open for reading")	9✔
141	with self._seek_lock():	9✔
142	self.on_access()	9✔
143	return self._bytes_io.read(size)	9✔
144
145	def readable(self):	9✔
146	# type: () -> bool
147	return self._mode.reading	9✔
148
149	def readinto(self, buffer):	9✔
150	# type (bytearray) -> Optional[int]
151	if not self._mode.reading:	9!
152	raise IOError("File not open for reading")	×
153	with self._seek_lock():	9✔
154	self.on_access()	9✔
155	return self._bytes_io.readinto(buffer)	9✔
156
157	def readlines(self, hint=-1):	9✔
158	# type: (int) -> List[bytes]
159	if not self._mode.reading:	9!
160	raise IOError("File not open for reading")	×
161	with self._seek_lock():	9✔
162	self.on_access()	9✔
163	return self._bytes_io.readlines(hint)	9✔
164
165	def seekable(self):	9✔
166	# type: () -> bool
167	return True	9✔
168
169	def seek(self, pos, whence=Seek.set):	9✔
170	# type: (int, SupportsInt) -> int
171	# NOTE(@althonos): allows passing both Seek.set and os.SEEK_SET
172	with self._seek_lock():	9✔
173	self.on_access()	9✔
174	return self._bytes_io.seek(pos, int(whence))	9✔
175
176	def tell(self):	9✔
177	# type: () -> int
178	return self.pos	9✔
179
180	def truncate(self, size=None):	9✔
181	# type: (Optional[int]) -> int
182	with self._seek_lock():	9✔
183	self.on_modify()	9✔
184	new_size = self._bytes_io.truncate(size)	9✔
185	if size is not None and self._bytes_io.tell() < size:	9✔
186	file_size = self._bytes_io.seek(0, os.SEEK_END)	9✔
187	self._bytes_io.write(b"\0" * (size - file_size))	9✔
188	self._bytes_io.seek(-size + file_size, os.SEEK_END)	9✔
189	return size or new_size	9✔
190
191	def writable(self):	9✔
192	# type: () -> bool
193	return self._mode.writing	9✔
194
195	def write(self, data):	9✔
196	# type: (Union[bytes, memoryview, array.array[Any], mmap.mmap]) -> int
197	if not self._mode.writing:	9✔
198	raise IOError("File not open for writing")	9✔
199	with self._seek_lock():	9✔
200	self.on_modify()	9✔
201	return self._bytes_io.write(data)	9✔
202
203	def writelines(self, sequence):	9✔
204	# type: (Iterable[Union[bytes, memoryview, array.array[Any], mmap.mmap]]) -> None # noqa: E501
205	with self._seek_lock():	9✔
206	self.on_modify()	9✔
207	self._bytes_io.writelines(sequence)	9✔
208
209
210	class _DirEntry(object):	9✔
211	def __init__(self, resource_type, name):	9✔
212	# type: (ResourceType, Text) -> None
213	self.resource_type = resource_type	9✔
214	self.name = name	9✔
215	self._dir = OrderedDict() # type: typing.MutableMapping[Text, _DirEntry]	9✔
216	self._open_files = [] # type: typing.MutableSequence[_MemoryFile]	9✔
217	self._bytes_file = None # type: Optional[io.BytesIO]	9✔
218	self.lock = RLock()	9✔
219
220	current_time = time.time()	9✔
221	self.created_time = current_time	9✔
222	self.accessed_time = current_time	9✔
223	self.modified_time = current_time	9✔
224
225	if not self.is_dir:	9✔
226	self._bytes_file = io.BytesIO()	9✔
227
228	@property	9✔
229	def bytes_file(self):	2✔
230	# type: () -> Optional[io.BytesIO]
231	return self._bytes_file	9✔
232
233	@property	9✔
234	def is_dir(self):	2✔
235	# type: () -> bool
236	return self.resource_type == ResourceType.directory	9✔
237
238	@property	9✔
239	def size(self):	2✔
240	# type: () -> int
241	with self.lock:	9✔
242	if self.is_dir:	9✔
243	return 0	9✔
244	else:
245	_bytes_file = typing.cast(io.BytesIO, self._bytes_file)	9✔
246	_bytes_file.seek(0, os.SEEK_END)	9✔
247	return _bytes_file.tell()	9✔
248
249	@overload
250	def get_entry(self, name, default): # noqa: F811
251	# type: (Text, _DirEntry) -> _DirEntry
252	pass
253
254	@overload
255	def get_entry(self, name): # noqa: F811
256	# type: (Text) -> Optional[_DirEntry]
257	pass
258
259	@overload
260	def get_entry(self, name, default): # noqa: F811
261	# type: (Text, None) -> Optional[_DirEntry]
262	pass
263
264	def get_entry(self, name, default=None): # noqa: F811	9✔
265	# type: (Text, Optional[_DirEntry]) -> Optional[_DirEntry]
266	assert self.is_dir, "must be a directory"	9✔
267	return self._dir.get(name, default)	9✔
268
269	def set_entry(self, name, dir_entry):	9✔
270	# type: (Text, _DirEntry) -> None
271	self._dir[name] = dir_entry	9✔
272
273	def remove_entry(self, name):	9✔
274	# type: (Text) -> None
275	del self._dir[name]	9✔
276
277	def __contains__(self, name):	9✔
278	# type: (object) -> bool
279	return name in self._dir	9✔
280
281	def __len__(self):	9✔
282	# type: () -> int
283	return len(self._dir)	9✔
284
285	def list(self):	9✔
286	# type: () -> List[Text]
287	return list(self._dir.keys())	9✔
288
289	def add_open_file(self, memory_file):	9✔
290	# type: (_MemoryFile) -> None
291	self._open_files.append(memory_file)	9✔
292
293	def remove_open_file(self, memory_file):	9✔
294	# type: (_MemoryFile) -> None
295	self._open_files.remove(memory_file)	9✔
296
297
298	@six.python_2_unicode_compatible	9✔
299	class MemoryFS(FS):	9✔
300	"""A filesystem that stored in memory.
301
302	Memory filesystems are useful for caches, temporary data stores,
303	unit testing, etc. Since all the data is in memory, they are very
304	fast, but non-permanent. The `MemoryFS` constructor takes no
305	arguments.
306
307	Examples:
308	Create with the constructor::
309
310	>>> from fs.memoryfs import MemoryFS
311	>>> mem_fs = MemoryFS()
312
313	Or via an FS URL::
314
315	>>> import fs
316	>>> mem_fs = fs.open_fs('mem://')
317
318	"""
319
320	_meta = {	9✔
321	"case_insensitive": False,
322	"invalid_path_chars": "\0",
323	"network": False,
324	"read_only": False,
325	"thread_safe": True,
326	"unicode_paths": True,
327	"virtual": False,
328	} # type: Dict[Text, Union[Text, int, bool, None]]
329
330	def __init__(self):	9✔
331	# type: () -> None
332	"""Create an in-memory filesystem."""
333	self._meta = self._meta.copy()	9✔
334	self.root = self._make_dir_entry(ResourceType.directory, "")	9✔
335	super(MemoryFS, self).__init__()	9✔
336
337	def __repr__(self):	9✔
338	# type: () -> str
339	return "MemoryFS()"	9✔
340
341	def __str__(self):	9✔
342	# type: () -> str
343	return "<memfs>"	9✔
344
345	def _make_dir_entry(self, resource_type, name):	9✔
346	# type: (ResourceType, Text) -> _DirEntry
347	return _DirEntry(resource_type, name)	9✔
348
349	def _get_dir_entry(self, dir_path):	9✔
350	# type: (Text) -> Optional[_DirEntry]
351	"""Get a directory entry, or `None` if one doesn't exist."""
352	with self._lock:	9✔
353	dir_path = normpath(dir_path)	9✔
354	current_entry = self.root # type: Optional[_DirEntry]	9✔
355	for path_component in iteratepath(dir_path):	9✔
356	if current_entry is None:	9✔
357	return None	9✔
358	if not current_entry.is_dir:	9✔
359	return None	9✔
360	current_entry = current_entry.get_entry(path_component)	9✔
361	return current_entry	9✔
362
363	def close(self):	9✔
364	# type: () -> None
365	if not self._closed:	9✔
366	del self.root	9✔
367	return super(MemoryFS, self).close()	9✔
368
369	def getinfo(self, path, namespaces=None):	9✔
370	# type: (Text, Optional[Collection[Text]]) -> Info
371	namespaces = namespaces or ()	9✔
372	_path = self.validatepath(path)	9✔
373	dir_entry = self._get_dir_entry(_path)	9✔
374	if dir_entry is None:	9✔
375	raise errors.ResourceNotFound(path)	9✔
376	info = {"basic": {"name": dir_entry.name, "is_dir": dir_entry.is_dir}}	9✔
377	if "details" in namespaces:	9✔
378	info["details"] = {	9✔
379	"_write": ["accessed", "modified"],
380	"type": int(dir_entry.resource_type),
381	"size": dir_entry.size,
382	"accessed": dir_entry.accessed_time,
383	"modified": dir_entry.modified_time,
384	"created": dir_entry.created_time,
385	}
386	return Info(info)	9✔
387
388	def listdir(self, path):	9✔
389	# type: (Text) -> List[Text]
390	self.check()	9✔
391	_path = self.validatepath(path)	9✔
392	with self._lock:	9✔
393	dir_entry = self._get_dir_entry(_path)	9✔
394	if dir_entry is None:	9✔
395	raise errors.ResourceNotFound(path)	9✔
396	if not dir_entry.is_dir:	9✔
397	raise errors.DirectoryExpected(path)	9✔
398	return dir_entry.list()	9✔
399
400	if typing.TYPE_CHECKING:	9!
401
402	def opendir(self, path, factory=None):	×
403	# type: (_M, Text, Optional[_OpendirFactory]) -> SubFS[_M]
404	pass	×
405
406	def makedir(	9✔
407	self, # type: _M
408	path, # type: Text
409	permissions=None, # type: Optional[Permissions]
410	recreate=False, # type: bool
411	):
412	# type: (...) -> SubFS[_M]
413	_path = self.validatepath(path)	9✔
414	with self._lock:	9✔
415	if _path == "/":	9✔
416	if recreate:	9✔
417	return self.opendir(path)	9✔
418	else:
419	raise errors.DirectoryExists(path)	9✔
420
421	dir_path, dir_name = split(_path)	9✔
422
423	parent_dir = self._get_dir_entry(dir_path)	9✔
424	if parent_dir is None:	9✔
425	raise errors.ResourceNotFound(path)	9✔
426
427	dir_entry = parent_dir.get_entry(dir_name)	9✔
428	if dir_entry is not None and not recreate:	9✔
429	raise errors.DirectoryExists(path)	9✔
430
431	if dir_entry is None:	9✔
432	new_dir = self._make_dir_entry(ResourceType.directory, dir_name)	9✔
433	parent_dir.set_entry(dir_name, new_dir)	9✔
434	return self.opendir(path)	9✔
435
436	def openbin(self, path, mode="r", buffering=-1, **options):	9✔
437	# type: (Text, Text, int, **Any) -> BinaryIO
438	_mode = Mode(mode)	9✔
439	_mode.validate_bin()	9✔
440	_path = self.validatepath(path)	9✔
441	dir_path, file_name = split(_path)	9✔
442
443	if not file_name:	9✔
444	raise errors.FileExpected(path)	9✔
445
446	with self._lock:	9✔
447	parent_dir_entry = self._get_dir_entry(dir_path)	9✔
448	if parent_dir_entry is None or not parent_dir_entry.is_dir:	9✔
449	raise errors.ResourceNotFound(path)	9✔
450
451	if _mode.create:	9✔
452	if file_name not in parent_dir_entry:	9✔
453	file_dir_entry = self._make_dir_entry(ResourceType.file, file_name)	9✔
454	parent_dir_entry.set_entry(file_name, file_dir_entry)	9✔
455	else:
456	file_dir_entry = self._get_dir_entry(_path) # type: ignore	9✔
457	if _mode.exclusive:	9✔
458	raise errors.FileExists(path)	9✔
459
460	if file_dir_entry.is_dir:	9✔
461	raise errors.FileExpected(path)	9✔
462
463	mem_file = _MemoryFile(	9✔
464	path=_path, memory_fs=self, mode=mode, dir_entry=file_dir_entry
465	)
466
467	file_dir_entry.add_open_file(mem_file)	9✔
468	return mem_file # type: ignore	9✔
469
470	if file_name not in parent_dir_entry:	9✔
471	raise errors.ResourceNotFound(path)	9✔
472
473	file_dir_entry = parent_dir_entry.get_entry(file_name) # type: ignore	9✔
474	if file_dir_entry.is_dir:	9✔
475	raise errors.FileExpected(path)	9✔
476
477	mem_file = _MemoryFile(	9✔
478	path=_path, memory_fs=self, mode=mode, dir_entry=file_dir_entry
479	)
480	file_dir_entry.add_open_file(mem_file)	9✔
481	return mem_file # type: ignore	9✔
482
483	def remove(self, path):	9✔
484	# type: (Text) -> None
485	_path = self.validatepath(path)	9✔
486
487	with self._lock:	9✔
488	dir_path, file_name = split(_path)	9✔
489	parent_dir_entry = self._get_dir_entry(dir_path)	9✔
490
491	if parent_dir_entry is None or file_name not in parent_dir_entry:	9✔
492	raise errors.ResourceNotFound(path)	9✔
493
494	file_dir_entry = typing.cast(_DirEntry, self._get_dir_entry(_path))	9✔
495	if file_dir_entry.is_dir:	9✔
496	raise errors.FileExpected(path)	9✔
497
498	parent_dir_entry.remove_entry(file_name)	9✔
499
500	def removedir(self, path):	9✔
501	# type: (Text) -> None
502	_path = self.validatepath(path)	9✔
503
504	if _path == "/":	9✔
505	raise errors.RemoveRootError()	9✔
506
507	with self._lock:	9✔
508	dir_path, file_name = split(_path)	9✔
509	parent_dir_entry = self._get_dir_entry(dir_path)	9✔
510
511	if parent_dir_entry is None or file_name not in parent_dir_entry:	9✔
512	raise errors.ResourceNotFound(path)	9✔
513
514	dir_dir_entry = typing.cast(_DirEntry, self._get_dir_entry(_path))	9✔
515	if not dir_dir_entry.is_dir:	9✔
516	raise errors.DirectoryExpected(path)	9✔
517
518	if len(dir_dir_entry):	9✔
519	raise errors.DirectoryNotEmpty(path)	9✔
520
521	parent_dir_entry.remove_entry(file_name)	9✔
522
523	def setinfo(self, path, info):	9✔
524	# type: (Text, RawInfo) -> None
525	_path = self.validatepath(path)	9✔
526	with self._lock:	9✔
527	dir_path, file_name = split(_path)	9✔
528	parent_dir_entry = self._get_dir_entry(dir_path)	9✔
529
530	if parent_dir_entry is None or file_name not in parent_dir_entry:	9✔
531	raise errors.ResourceNotFound(path)	9✔
532
533	resource_entry = typing.cast(	9✔
534	_DirEntry, parent_dir_entry.get_entry(file_name)
535	)
536
537	if "details" in info:	9!
538	details = info["details"]	9✔
539	if "accessed" in details:	9!
540	resource_entry.accessed_time = details["accessed"] # type: ignore	9✔
541	if "modified" in details:	9!
542	resource_entry.modified_time = details["modified"] # type: ignore	9✔

PyFilesystem / pyfilesystem2 / 518483918

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous