patch Python builtins with PyFilesystem

fs/patcher/__init__.py

@@ -0,0 +1 @@
+from ._patch import patch

fs/patcher/__main__.py

@@ -0,0 +1,24 @@
+from __future__ import unicode_literals
+
+from fs import open_fs
+from . import patch
+
+import os
+from os import listdir
+from os.path import exists
+fs = open_fs('mem://')
+foo = fs.makedir('foo')
+foo.touch('bar1')
+foo.touch('bar2')
+with patch(fs):
+    d = os.listdir('/')
+    print(d)
+    print(listdir('/'))
+    # print('--')
+    # print(os.path.exists('foo'))
+    # print(exists('foo'))
+    # print('--')
+    # print(os.path.isfile('foo'))
+    # print(os.path.getatime('foo'))
+
+print(os.listdir('/'))

fs/patcher/_patch.py

@@ -0,0 +1,53 @@
+from __future__ import unicode_literals
+
+from six import PY3
+
+from .context import PatchContext
+from . import common_patches
+from ..opener import open_fs
+from ..base import FS
+
+
+PATCHES = [
+    # common_patches.OsPathExists(),
+    # common_patches.OsPathLExists(),
+    # common_patches.OsPathGetatime(),
+    # common_patches.OsPathGetmtime(),
+    # common_patches.OsPathGetctime(),
+    # common_patches.OsPathGetsize(),
+    # common_patches.OsPathIsfile(),
+    # common_patches.OsPathIsdir(),
+    # common_patches.OsPathIslink(),
+    common_patches.OsListdir(),
+]
+
+if PY3:
+    from . import py3_patches
+
+
+
+def patch(fs_url, patches=None, cwd='/'):
+    """
+    Patch a filesystem over Python builtins.
+
+    """
+
+    if isinstance(fs_url, FS):
+        filesystem = fs_url
+        auto_close = True
+    else:
+        filesystem = open_fs(fs_url)
+        auto_close = False
+
+    _patches = (
+        patches
+        if patches is not None
+        else PATCHES
+    )
+
+    return PatchContext(
+        filesystem,
+        PATCHES,
+        cwd=cwd,
+        auto_close=auto_close
+    )

fs/patcher/common_patches.py

@@ -0,0 +1,198 @@
+from __future__ import unicode_literals
+
+import builtins
+import errno
+import os
+import sys
+from operator import attrgetter
+
+from six import text_type
+
+from ..path import join
+from ..errors import FSError, ResourceNotFound
+from .context import ModulePatch, CodePatch
+
+from . import stack
+
+
+def _not_found(path):
+    """Raise an OSError ENOENT error."""
+    _error = os.strerror(errno.ENOENT)
+    raise OSError(errno.ENOENT, _error, path)
+
+
+from fs.patcher.stack import get_context
+
+class OsPathExists(CodePatch):
+    module = os.path
+    attrib = 'exists'
+
+    def exists(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        return context.filesystem.exists(_path)
+
+
+class OsPathLExists(CodePatch):
+    module = os.path
+    attrib = 'lexists'
+
+    def lexists(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        return context.filesystem.exists(_path)
+
+
+class OsPathGetatime(CodePatch):
+    module = os.path
+    attrib = 'getatime'
+
+    def getatime(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        try:
+            info = context.filesystem.getdetails(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+        try:
+            _time = info.raw['details']['accessed']
+        except KeyError:
+            raise OSError('unable to get time')
+        return _time
+
+
+class OsPathGetmtime(CodePatch):
+    module = os.path
+    attrib = 'getmtime'
+
+    def getmtime(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        try:
+            info = context.filesystem.getdetails(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+        try:
+            _time = info.raw['details']['modified']
+        except KeyError:
+            raise OSError('unable to get time')
+        return _time
+
+
+class OsPathGetctime(CodePatch):
+    module = os.path
+    attrib = 'getctime'
+
+    def getctime(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        try:
+            info = context.filesystem.getdetails(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+        time_field = (
+            'metadata_changed'
+            if sys.platform == 'linux' else
+            'created'
+        )
+        try:
+            _time = info.raw['details'][time_field]
+        except KeyError:
+            raise OSError('unable to get time')
+        return _time
+
+
+class OsPathGetsize(CodePatch):
+    module = os.path
+    attrib = 'getsize'
+
+    def getsize(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        try:
+            return context.filesystem.getsize(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+
+
+class OsPathIsfile(CodePatch):
+    module = os.path
+    attrib = 'isfile'
+
+    def isfile(path):
+        from fs.patcher.stack import get_context
+        from fs.errors import ResourceNotFound
+        context = get_context()
+        _path = context.get_path(path)
+        try:
+            return context.filesystem.isfile(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+
+
+class OsPathIsdir(CodePatch):
+    module = os.path
+    attrib = 'isdir'
+
+    def isdir(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = context.get_path(path)
+        try:
+            return context.filesystem.isdir(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+
+
+class OsPathIslink(CodePatch):
+    module = os.path
+    attrib = 'islink'
+
+    def islink(path):
+        from fs.patcher.stack import get_context
+        context = get_context()
+        _path = self.context.get_path(path)
+        try:
+            return self.context.filesystem.islink(_path)
+        except ResourceNotFound:
+            _not_found(_path)
+
+
+class OsListdir(ModulePatch):
+    module = os
+    attrib = 'listdir'
+
+    def listdir(self, path='.'):
+        if not path:
+            _not_found(path)
+        _path = self.context.get_path(path)
+        try:
+            dirlist = self.context.filesystem.listdir(_path)
+        except ResourceNotFound as error:
+            _not_found(path)
+        except FSError as error:
+            raise OSError(text_type(error))
+        return dirlist
+
+
+class OsWalk(ModulePatch):
+    module = os
+    attrib = 'walk'
+
+    def walk(self, top, topdown=True, onerror=None, followlinks=False):
+        _path = self.context.get_path(top)
+        walk_method = 'depth' if topdown else 'breadth'
+        getname = attrgetter('name')
+        with self.context.filesystem.opendir(_path) as dir_fs:
+            for _dirpath, _dirs, _files in dir_fs.walk(method=walk_method):
+                dirpath = join(_path, _dirpath)
+                dirs = [getname(info) for info in _dirs]
+                files = [getname(info) for info in _files]
+                yield dirpath, dirs, files
+

fs/patcher/context.py

@@ -0,0 +1,119 @@
+from __future__ import unicode_literals
+
+from functools import partial
+
+import six
+
+from .stack import context_stack
+from ..path import join
+
+
+
+class Patch(object):
+    """Applies a individual patch."""
+
+    def apply(self, context):
+        """Apply the patch."""
+        raise NotImplementedError
+
+    def revert(self, context):
+        """Restore to previous state."""
+        raise NotImplementedError
+
+
+class ModulePatch(Patch):
+    module = None
+    attrib = None
+    method_name = None
+
+    def __init__(self):
+        assert self.module is not None, "module must be a classvar"
+        self._restore = None
+        self.context = None
+
+    def __repr__(self):
+        return '<{} {}.{}>'.format(
+            self.__class__.__name__,
+            self.module.__name__,
+            self.attrib
+        )
+
+    @property
+    def replace_method(self):
+        return getattr(self, (self.method_name or self.attrib))
+
+    def apply(self, context):
+        self.context = context
+        self._restore = getattr(self.module, self.attrib)
+        setattr(self.module, self.attrib, self.replace_method)
+
+    def revert(self, context):
+        setattr(self.module, self.attrib, self._restore)
+        self._restore = None
+
+
+class CodePatch(ModulePatch):
+    module = None
+    attrib = None
+
+    def apply(self, context):
+        self.context = context
+        original_method = getattr(self.module, self.attrib)
+
+        self._restore = getattr(
+            original_method,
+            '__code__'
+        )
+        patch = getattr(self, (self.method_name or self.attrib))
+        replace_code = patch.__code__
+        setattr(
+            original_method,
+            '__code__',
+            replace_code
+        )
+
+    def revert(self, context):
+        setattr(
+            getattr(self.module, self.attrib),
+            '__code__',
+            self._restore
+        )
+        self._restore = None
+        self.context = None
+
+
+class PatchContext(object):
+    """
+    Patches a filesystem over builtins.
+
+    """
+    def __init__(self, filesystem, patches, cwd='/', auto_close=False):
+        self.filesystem = filesystem
+        self.patches = patches
+        self.cwd = cwd
+        self.auto_close = auto_close
+
+        self.file_handles = {}
+        self._applied_patches = []
+
+    def get_path(self, path):
+        path = join(self.cwd, path)
+        return path
+
+    def __enter__(self):
+        context_stack.append(self)
+        for patch in self.patches:
+            patch.apply(self)
+            self._applied_patches.append(patch)
+        return self
+
+    def __exit__(self, exc_type, exc_value, tb):
+        context_stack.pop()
+        pop = self._applied_patches.pop
+        try:
+            while self._applied_patches:
+                patch = pop()
+                patch.revert(self)
+        finally:
+            if self.auto_close:
+                self.filesystem.close()