View on GitHub


25 mins
Test Coverage

import os
import shutil
from contextlib import contextmanager

from salve.filesys.abstract import Filesys
from salve.util import hash_from_path

class ConcreteFilesys(Filesys):
    def lookup_type(self, path):
        Lookup the type of a given path.

            The path to the file, dir, or link to lookup.
        # if no constructor is given, inspect the underlying filesystem
        if os.path.islink(path):
            return self.element_types.LINK
        elif os.path.isdir(path):
            return self.element_types.DIR
        elif os.path.isfile(path):
            return self.element_types.FILE
            return None

    def access(self, path, mode):
        Transparent implementation of access() using os.access

            The type of access being inspected
        return os.access(path, mode)

    def stat(self, path):
        Transparent implementation of stat() using os.lstat
        return os.lstat(path)

    def chmod(self, path, *args, **kwargs):
        Transparent implementation of chmod() using os.chmod
        return os.chmod(path, *args, **kwargs)

    def chown(self, path, uid, gid):
        Transparent implementation of chown() using os.lchown
        return os.lchown(path, uid, gid)

    def exists(self, path):
        Transparent implementation of exists() using os.path.lexists
        return os.path.lexists(path)

    def hash(self, path):
        Transparent implementation of hash() using
        assert self.exists(path)
        return hash_from_path(path)

    def copy(self, src, dst):
        Copies the source to the destination on the underlying filesys. Does
        not necessarily create a new entry registered to the destination.

            The origin path to be copied. When looked up in the filesys, must
            not be None and must satisfy exists()

            The destination path for the copy operation. The ancestors of this
            path must exist so that the file creation will not fail.
        assert self.exists(src)

        src_ty = self.lookup_type(src)
        if src_ty == self.element_types.FILE:
            shutil.copyfile(src, dst)
        # FIXME: copytree is kind of scary. Eventually would like to replace
        # this with a NotImplementedError in order to force explicit creates
        # and file copies
        elif src_ty == self.element_types.DIR:
            shutil.copytree(src, dst, symlinks=True)
        elif src_ty == self.element_types.LINK:
            os.symlink(os.readlink(src), dst)
        else:  # pragma: no cover
            assert False

    def open(self, path, *args, **kwargs):
        Transparent implementation of open() using builtin open
        with open(path, *args, **kwargs) as f:
            yield f

    def touch(self, path, *args, **kwargs):
        Touch a file by opening it in append mode and closing it
        with, 'a'):

    def symlink(self, path, target):
        Transparent implementation of symlink() using os.symlink
        os.symlink(path, target)

    def walk(self, path, *args, **kwargs):
        Transparent implementation of walk() using os.walk
        for x in os.walk(path, *args, **kwargs):
            yield x

    def mkdir(self, path, recursive=True):
        Use os.mkdir or os.makedirs to create the directory desired,
        suppressing "already exists" errors.
        May still return OSError(2, 'No such file or directory') when
        nonrecursive mkdir calls have missing ancestors.
            if recursive:
        except OSError as e:
            # 'File exists' errno
            if e.errno == 17:
                raise e

    def get_existing_ancestor(self, path):
        Finds the longest prefix to a path that is known to exist.

            An absolute path whose prefix should be inspected.
        # be safe
        path = os.path.abspath(path)
        # exists is sufficient because we can stat directories as long as
        # we have execute permissions
        while not os.path.exists(path):
            path = os.path.dirname(path)
        return path