Skip to content

P
pyfs
Commit 563213f3 by willmcgugan
Options

Tidied up the code, and added an ObjectTree class

parent 63ccf2cd
Showing with 200 additions and 202 deletions
fs/fs.py
......@@ -10,18 +10,31 @@ import datetime
error_msgs = {
"UNKNOWN_ERROR" : "No information on error: %(path)s",
# UnsupportedError
"UNSUPPORTED" : "Action is unsupported by this filesystem.",
"INVALID_PATH" : "Path is invalid: %(path)s",
"NO_DIR" : "Directory does not exist: %(path)s",
"NO_FILE" : "No such file: %(path)s",
"NO_RESOURCE" : "No path to: %(path)s",
# OperationFailedError
"LISTDIR_FAILED" : "Unable to get directory listing: %(path)s",
"DELETE_FAILED" : "Unable to delete file: %(path)s",
"RENAME_FAILED" : "Unable to rename file: %(path)s",
"NO_SYS_PATH" : "No mapping to OS filesytem: %(path)s,",
"DIR_EXISTS" : "Directory exists (try allow_recreate=True): %(path)s",
"OPEN_FAILED" : "Unable to open file: %(path)s",
"DIR_EXISTS" : "Directory exists (try allow_recreate=True): %(path)s",
# NoSysPathError
"NO_SYS_PATH" : "No mapping to OS filesytem: %(path)s,",
# PathError
"INVALID_PATH" : "Path is invalid: %(path)s",
# ResourceLockedError
"FILE_LOCKED" : "File is locked: %(path)s",
"DIR_LOCKED" : "Dir is locked: %(path)s",
# ResourceNotFoundError
"NO_DIR" : "Directory does not exist: %(path)s",
"NO_FILE" : "No such file: %(path)s",
"NO_RESOURCE" : "No path to: %(path)s",
}
error_codes = error_msgs.keys()
......@@ -31,7 +44,6 @@ class FSError(Exception):
"""A catch all exception for FS objects."""
def __init__(self, code, path=None, msg=None, details=None):
"""
code -- A short identifier for the error
......@@ -47,18 +59,27 @@ class FSError(Exception):
self.details = details
def __str__(self):
msg = self.msg % dict((k, str(v)) for k, v in self.__dict__.iteritems())
return '%s. %s' % (self.code, msg)
class PathError(Exception):
class UnsupportedError(FSError):
pass
def __init__(self, msg):
self.msg = msg
class OperationFailedError(FSError):
pass
def __str__(self):
return self.msg
class NoSysPathError(FSError):
pass
class PathError(FSError):
pass
class ResourceLockedError(FSError):
pass
class ResourceNotFoundError(FSError):
pass
class NullFile(object):
......@@ -143,7 +164,7 @@ def pathjoin(*paths):
for component in chain(*(normpath(path).split('/') for path in relpaths)):
if component == "..":
if not pathstack:
raise PathError("relative path is invalid")
raise PathError("INVALID_PATH", repr(paths), msg="relative path is invalid")
sub = pathstack.pop()
elif component == ".":
pass
......@@ -187,9 +208,7 @@ def makeabsolute(path):
return path
def _iteratepath(path, numsplits=None):
path = resolvepath(path)
if not path:
return []
......@@ -200,9 +219,7 @@ def _iteratepath(path, numsplits=None):
def print_fs(fs, path="/", max_levels=None, indent=' '*2):
def print_dir(fs, path, level):
try:
dir_listing = [(fs.isdir(pathjoin(path,p)), p) for p in fs.listdir(path)]
except FSError, e:
......@@ -219,7 +236,6 @@ def print_fs(fs, path="/", max_levels=None, indent=' '*2):
print_dir(fs, pathjoin(path, item), level+1)
else:
print indent*level + '%s' % item
print_dir(fs, path, 0)
......@@ -227,13 +243,10 @@ class FS(object):
def _resolve(self, pathname):
resolved_path = resolvepath(pathname)
return resolved_path
def _abspath(self, pathname):
pathname = normpath(pathname)
if not pathname.startswith('/'):
......@@ -241,21 +254,49 @@ class FS(object):
return pathname
def getsyspath(self, path):
raise NoSysPathError("NO_SYS_PATH", path)
raise FSError("NO_SYS_PATH", path)
def open(self, path, mode="r", buffering=-1, **kwargs):
raise UnsupportedError("UNSUPPORTED")
def safeopen(self, *args, **kwargs):
try:
f = self.open(*args, **kwargs)
except FSError, e:
if e.code == "NO_FILE":
except ResourceNotFoundError:
return NullFile()
raise
return f
def desc(self, path):
def exists(self, path):
raise UnsupportedError("UNSUPPORTED")
def isdir(self, path):
raise UnsupportedError("UNSUPPORTED")
def isfile(self, path):
raise UnsupportedError("UNSUPPORTED")
def ishidden(self, path):
return path.startswith('.')
def listdir(self, path="./", wildcard=None, full=False, absolute=False, hidden=False, dirs_only=False, files_only=False):
raise UnsupportedError("UNSUPPORTED")
def mkdir(self, path, mode=0777, recursive=False):
raise UnsupportedError("UNSUPPORTED")
def remove(self, path):
raise UnsupportedError("UNSUPPORTED")
def removedir(self, path, recursive=False):
raise UnsupportedError("UNSUPPORTED")
def rename(self, src, dst):
raise UnsupportedError("UNSUPPORTED")
def getinfo(self, path):
raise UnsupportedError("UNSUPPORTED")
def desc(self, path):
if not self.exists(path):
return "No description available"
......@@ -270,29 +311,19 @@ class FS(object):
return "OS file, maps to %s" % sys_path
def open(self, path, mode="r", buffering=-1, **kwargs):
raise FSError("UNSUPPORTED")
raise UNSUPPORTED_ERROR("UNSUPPORTED")
def opendir(self, path):
if not self.exists(path):
raise FSError("NO_DIR", path)
raise ResourceNotFoundError("NO_DIR", path)
sub_fs = SubFS(self, path)
return sub_fs
def remove(self, path):
raise FSError("UNSUPPORTED", path)
def _listdir_helper(self, path, paths, wildcard, full, absolute, hidden, dirs_only, files_only):
if dirs_only and files_only:
raise ValueError("dirs_only and files_only can not both be True")
if wildcard is not None:
match = fnmatch.fnmatch
paths = [p for p in path if match(p, wildcard)]
......@@ -313,8 +344,7 @@ class FS(object):
return paths
def walk_files(self, path="/", wildcard=None, dir_wildcard=None):
def walkfiles(self, path="/", wildcard=None, dir_wildcard=None):
dirs = [path]
files = []
......@@ -337,12 +367,8 @@ class FS(object):
yield path
def walk(self, path="/", wildcard=None, dir_wildcard=None):
dirs = [path]
while dirs:
current_path = dirs.pop()
paths = []
......@@ -360,73 +386,58 @@ class FS(object):
paths.append(path)
else:
paths.append(path)
yield (current_path, paths)
def getsize(self, path):
return self.getinfo(path)['size']
class SubFS(FS):
def __init__(self, parent, sub_dir):
self.parent = parent
self.sub_dir = parent._abspath(sub_dir)
def __str__(self):
return "<SubFS \"%s\" in %s>" % (self.sub_dir, self.parent)
def desc(self, path):
if self.isdir(path):
return "Sub dir of %s"%str(self.parent)
else:
return "File in sub dir of %s"%str(self.parent)
def _delegate(self, path):
return pathjoin(self.sub_dir, resolvepath(makerelative(path)))
def getsyspath(self, path):
return self.parent.getsyspath(self._delegate(path))
def open(self, path, mode="r", buffering=-1, **kwargs):
return self.parent.open(self._delegate(path), mode, buffering)
def exists(self, path):
return self.parent.exists(self._delegate(path))
def opendir(self, path):
if not self.exists(path):
raise FSError("NO_DIR", path)
raise ResourceNotFoundError("NO_DIR", path)
path = self._delegate(path)
sub_fs = self.parent.opendir(path)
return sub_fs
def isdir(self, path):
return self.parent.isdir(self._delegate(path))
def isfile(self, path):
return self.parent.isdir(self._delegate(path))
def ishidden(self, path):
return self.parent.ishidden(self._delegate(path))
def listdir(self, path="./", wildcard=None, full=False, absolute=False, hidden=False, dirs_only=False, files_only=False):
paths = self.parent.listdir(self._delegate(path),
wildcard,
False,
......@@ -444,30 +455,23 @@ class SubFS(FS):
def mkdir(self, path, mode=0777, recursive=False):
return self.parent.mkdir(self._delegate(path), mode=mode, recursive=False)
def remove(self, path):
return self.parent.remove(self._delegate(path))
def removedir(self, path, recursive=False):
self.parent.removedir(self._delegate(path), recursive=False)
def getinfo(self, path):
return self.parent.getinfo(self._delegate(path))
def getsize(self, path):
return self.parent.getsize(self._delegate(path))
def rename(self, src, dst):
return self.parent.rename(self._delegate(src), self._delegate(dst))
def validatefs(fs):
expected_methods = [ "abspath",
......
fs/memoryfs.py
......@@ -18,7 +18,6 @@ def _check_mode(mode, mode_chars):
class MemoryFile(object):
def __init__(self, path, memory_fs, value, mode):
self.path = path
self.memory_fs = memory_fs
self.mode = mode
......@@ -54,11 +53,9 @@ class MemoryFile(object):
self.closed = False
def __del__(self):
if not self.closed:
self.close()
def flush(self):
pass
......@@ -99,14 +96,11 @@ class MemoryFile(object):
return self.mem_file.writelines(*args, **kwargs)
class MemoryFS(FS):
class DirEntry(object):
def __init__(self, type, name, contents=None):
self.type = type
self.name = name
self.permissions = None
......@@ -148,11 +142,9 @@ class MemoryFS(FS):
return "%s: %s" % (self.name, self.desc_contents())
def _make_dir_entry(self, *args, **kwargs):
return self.dir_entry_factory(*args, **kwargs)
def __init__(self, file_factory=None):
self.dir_entry_factory = MemoryFS.DirEntry
self.file_factory = file_factory or MemoryFile
......@@ -162,9 +154,7 @@ class MemoryFS(FS):
return "<MemoryFS>"
def _get_dir_entry(self, dirpath):
current_dir = self.root
for path_component in _iteratepath(dirpath):
dir_entry = current_dir.contents.get(path_component, None)
if dir_entry is None:
......@@ -173,12 +163,7 @@ class MemoryFS(FS):
return current_dir
def getsyspath(self, pathname):
raise FSError("NO_SYS_PATH", pathname, msg="This file-system has no syspath")
def desc(self, path):
if self.isdir(path):
return "Memory dir"
elif self.isfile(path):
......@@ -187,25 +172,21 @@ class MemoryFS(FS):
return "No description available"
def isdir(self, path):
dir_item = self._get_dir_entry(self._resolve(path))
if dir_item is None:
return False
return dir_item.isdir()
def isfile(self, path):
dir_item = self._get_dir_entry(self._resolve(path))
if dir_item is None:
return False
return dir_item.isfile()
def exists(self, path):
return self._get_dir_entry(path) is not None
def mkdir(self, dirname, mode=0777, recursive=False, allow_recreate=False):
fullpath = dirname
dirpath, dirname = pathsplit(dirname)
......@@ -213,11 +194,11 @@ class MemoryFS(FS):
parent_dir = self._get_dir_entry(dirpath)
if parent_dir is not None:
if parent_dir.isfile():
raise FSError("NO_DIR", dirname, msg="Can not create a directory, because path references a file: %(path)s")
raise ResourceNotFoundError("NO_DIR", dirname, msg="Can not create a directory, because path references a file: %(path)s")
else:
if not allow_recreate:
if dirname in parent_dir.contents:
raise FSError("NO_DIR", dirname, msg="Can not create a directory that already exists (try allow_recreate=True): %(path)s")
raise ResourceNotFoundError("NO_DIR", dirname, msg="Can not create a directory that already exists (try allow_recreate=True): %(path)s")
current_dir = self.root
for path_component in _iteratepath(dirpath)[:-1]:
......@@ -225,7 +206,7 @@ class MemoryFS(FS):
if dir_item is None:
break
if not dir_item.isdir():
raise FSError("NO_DIR", dirname, msg="Can not create a directory, because path references a file: %(path)s")
raise ResourceNotFoundError("NO_DIR", dirname, msg="Can not create a directory, because path references a file: %(path)s")
current_dir = dir_item.contents
current_dir = self.root
......@@ -243,7 +224,7 @@ class MemoryFS(FS):
else:
parent_dir = self._get_dir_entry(dirpath)
if parent_dir is None:
raise FSError("NO_DIR", dirname, msg="Could not make dir, as parent dir does not exist: %(path)s")
raise ResourceNotFoundError("NO_DIR", dirname, msg="Could not make dir, as parent dir does not exist: %(path)s")
dir_item = parent_dir.contents.get(dirname, None)
if dir_item is not None:
......@@ -251,55 +232,46 @@ class MemoryFS(FS):
if not allow_recreate:
raise FSError("DIR_EXISTS", dirname)
else:
raise FSError("NO_DIR", dirname, msg="Can not create a directory, because path references a file: %(path)s")
raise ResourceNotFoundError("NO_DIR", dirname, msg="Can not create a directory, because path references a file: %(path)s")
if dir_item is None:
parent_dir.contents[dirname] = self._make_dir_entry("dir", dirname)
return self
def _lock_dir_entry(self, path):
dir_entry = self._get_dir_entry(path)
dir_entry.lock()
def _unlock_dir_entry(self, path):
dir_entry = self._get_dir_entry(path)
dir_entry.unlock()
def _is_dir_locked(self, path):
dir_entry = self._get_dir_entry(path)
return dir_entry.islocked()
def open(self, path, mode="r", **kwargs):
filepath, filename = pathsplit(path)
parent_dir_entry = self._get_dir_entry(filepath)
if parent_dir_entry is None or not parent_dir_entry.isdir():
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
if 'r' in mode or 'a' in mode:
if filename not in parent_dir_entry.contents:
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
file_dir_entry = parent_dir_entry.contents[filename]
if 'a' in mode and file_dir_entry.islocked():
raise FSError("FILE_LOCKED", path)
raise ResourceLockedError("FILE_LOCKED", path)
self._lock_dir_entry(path)
mem_file = self.file_factory(path, self, file_dir_entry.data, mode)
return mem_file
elif 'w' in mode:
if filename not in parent_dir_entry.contents:
file_dir_entry = self._make_dir_entry("file", filename)
parent_dir_entry.contents[filename] = file_dir_entry
......@@ -307,7 +279,7 @@ class MemoryFS(FS):
file_dir_entry = parent_dir_entry.contents[filename]
if file_dir_entry.islocked():
raise FSError("FILE_LOCKED", path)
raise ResourceLockedError("FILE_LOCKED", path)
self._lock_dir_entry(path)
......@@ -315,18 +287,16 @@ class MemoryFS(FS):
return mem_file
if parent_dir_entry is None:
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
def remove(self, path):
dir_entry = self._get_dir_entry(path)
if dir_entry is None:
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
if dir_entry.islocked():
raise FSError("FILE_LOCKED", path)
raise ResourceLockedError("FILE_LOCKED", path)
pathname, dirname = pathsplit(path)
......@@ -334,29 +304,21 @@ class MemoryFS(FS):
del parent_dir.contents[dirname]
def _on_close_memory_file(self, path, value):
filepath, filename = pathsplit(path)
dir_entry = self._get_dir_entry(path)
dir_entry.data = value
self._unlock_dir_entry(path)
def listdir(self, path="/", wildcard=None, full=False, absolute=False, hidden=False, dirs_only=False, files_only=False):
dir_entry = self._get_dir_entry(path)
if dir_entry is None:
raise FSError("NO_DIR", path)
raise ResourceNotFoundError("NO_DIR", path)
paths = dir_entry.contents.keys()
return self._listdir_helper(path, paths, wildcard, full, absolute, hidden, dirs_only, files_only)
def getinfo(self, path):
dir_entry = self._get_dir_entry(path)
info = {}
......
fs/mountfs.py
......@@ -22,14 +22,11 @@ class MountFS(FS):
else:
return dir_entry.data
def __init__(self):
self.mounts = {}
self.mem_fs = MemoryFS(file_factory=self.get_mount)
def _delegate(self, path):
path_components = list(_iteratepath(path))
current_dir = self.mem_fs.root
......@@ -57,18 +54,14 @@ class MountFS(FS):
return self, "", path
def desc(self, path):
fs, mount_path, delegate_path = self._delegate(path)
if fs is self:
return "Mount dir"
return "Mounted dir, maps to path %s on %s" % (delegate_path, str(fs))
def isdir(self, path):
fs, mount_path, delegate_path = self._delegate(path)
if fs is None:
return False
......@@ -79,11 +72,10 @@ class MountFS(FS):
return fs.isdir(delegate_path)
def listdir(self, path="/", wildcard=None, full=False, absolute=False, hidden=False, dirs_only=False, files_only=False):
fs, mount_path, delegate_path = self._delegate(path)
if fs is None:
raise FSError("NO_DIR", path)
raise ResourceNotFoundError("NO_DIR", path)
if fs is self:
if files_only:
......@@ -113,7 +105,6 @@ class MountFS(FS):
return paths
def mount(self, name, path, fs):
self.mem_fs.mkdir(path, recursive=True)
mount_filename = pathjoin(path, '.mount')
mount = self.mem_fs.open(mount_filename, 'w')
......
fs/multifs.py
......@@ -11,43 +11,34 @@ class MultiFS(FS):
self.fs_lookup = {}
def __str__(self):
return "<MultiFS: %s>" % ", ".join(str(fs) for fs in self.fs_sequence)
def addfs(self, name, fs):
if name in self.fs_lookup:
raise ValueError("Name already exists.")
self.fs_sequence.append(fs)
self.fs_lookup[name] = fs
def removefs(self, name):
fs = self.fs_lookup[name]
self.fs_sequence.remove(fs)
del self.fs_lookup[name]
def __getitem__(self, name):
return self.fs_lookup[name]
def __iter__(self):
return iter(self.fs_sequence)
def _delegate_search(self, path):
for fs in self:
if fs.exists(path):
return fs
return None
def which(self, path):
for fs in self:
if fs.exists(path):
for fs_name, fs_object in self.fs_lookup.iteritems():
......@@ -55,18 +46,14 @@ class MultiFS(FS):
return fs_name, fs
return None, None
def getsyspath(self, path):
fs = self._delegate_search(path)
if fs is not None:
return fs.getsyspath(path)
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
def desc(self, path):
if not self.exists(path):
raise FSError("NO_RESOURCE", path)
......@@ -78,41 +65,35 @@ class MultiFS(FS):
def open(self, path, mode="r", buffering=-1, **kwargs):
for fs in self:
if fs.exists(path):
fs_file = fs.open(path, mode, buffering, **kwargs)
return fs_file
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
def exists(self, path):
return self._delegate_search(path) is not None
def isdir(self, path):
fs = self._delegate_search(path)
if fs is not None:
return fs.isdir(path)
return False
def isfile(self, path):
fs = self._delegate_search(path)
if fs is not None:
return fs.isfile(path)
return False
def ishidden(self, path):
fs = self._delegate_search(path)
if fs is not None:
return fs.isfile(path)
return False
def listdir(self, path="./", *args, **kwargs):
paths = []
for fs in self:
try:
......@@ -123,29 +104,32 @@ class MultiFS(FS):
return list(set(paths))
def remove(self, path):
for fs in self:
if fs.exists(path):
fs.remove(path)
return
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
def removedir(self, path, recursive=False):
for fs in self:
if fs.isdir(path):
fs.removedir(path, recursive)
return
raise FSError("NO_DIR", path)
raise ResourceNotFoundError("NO_DIR", path)
def rename(self, src, dst):
for fs in self:
if fs.exists(src):
fs.rename(src, dst)
return
raise FSError("NO_RESOURCE", path)
def getinfo(self, path):
for fs in self:
if fs.exists(path):
return fs.getinfo(path)
raise FSError("NO_FILE", path)
raise ResourceNotFoundError("NO_FILE", path)
if __name__ == "__main__":
......
fs/objecttree.py 0 → 100644
#!/usr/bin/env python
from fs import _iteratepath, pathsplit
class ObjectDict(dict):
pass
class ObjectTree(object):
"""A class to facilitate the creation of tree structures."""
def __init__(self):
self.root = ObjectDict()
def _locate(self, path):
current = self.root
for path_component in path.split('/'):
if type(current) is not ObjectDict:
return None
node = current.get(path_component, None)
if node is None:
return None
current = node
return node
def __setitem__(self, path, object):
if not path:
raise IndexError("No path supplied")
current = self.root
path, name = path.rsplit('/', 1)
for path_component in path.split('/'):
node = current.get(path_component, None)
if node is None or type(node) is not ObjectDict:
new_dict = ObjectDict()
current[path_component] = new_dict
current = new_dict
else:
current = node
current[name] = object
def __getitem__(self, path):
node = self._locate(path)
if node is None:
raise IndexError("Path does not exist")
return node
def get(self, path, default):
node = self._locate(path)
if node is None:
return default
return node
def __contains__(self, value):
return value in self.root
def __iter__(self):
return iter(self.root)
def keys(self):
return self.root.keys()
def iterkeys(self):
return self.root.keys()
def items(self):
return self.root.items()
def iteritems(self):
return self.root.iteritems()
if __name__ == "__main__":
ot = ObjectTree()
ot['a/b/c'] = "Hai!"
print ot['a/b/c']
ot['a/b/c/d'] = "?"
print ot['a/b/c'].keys()
\ No newline at end of file
fs/osfs.py
......@@ -9,63 +9,51 @@ class OSFS(FS):
expanded_path = normpath(os.path.expanduser(os.path.expandvars(root_path)))
if not os.path.exists(expanded_path):
raise FSError("NO_DIR", expanded_path, msg="Root directory does not exist: %(path)s")
raise ResourceNotFoundError("NO_DIR", expanded_path, msg="Root directory does not exist: %(path)s")
if not os.path.isdir(expanded_path):
raise FSError("NO_DIR", expanded_path, msg="Root path is not a directory: %(path)s")
raise ResourceNotFoundError("NO_DIR", expanded_path, msg="Root path is not a directory: %(path)s")
self.root_path = normpath(os.path.abspath(expanded_path))
def __str__(self):
return "<OSFS \"%s\">" % self.root_path
def getsyspath(self, pathname):
sys_path = os.path.join(self.root_path, makerelative(self._resolve(pathname)))
def getsyspath(self, path):
sys_path = os.path.join(self.root_path, makerelative(self._resolve(path)))
return sys_path
def open(self, pathname, mode="r", buffering=-1, **kwargs):
def open(self, path, mode="r", buffering=-1, **kwargs):
try:
f = open(self.getsyspath(pathname), mode, buffering)
f = open(self.getsyspath(path), mode, buffering)
except IOError, e:
raise FSError("OPEN_FAILED", pathname, details=e, msg=str(details))
raise OperationFailedError("OPEN_FAILED", path, details=e, msg=str(details))
return f
def exists(self, pathname):
def exists(self, path):
path = self.getsyspath(path)
return os.path.exists(path)
pathname = self.getsyspath(pathname)
return os.path.exists(pathname)
def isdir(self, pathname):
path = self.getsyspath(pathname)
def isdir(self, path):
path = self.getsyspath(path)
return os.path.isdir(path)
def isfile(self, pathname):
path = self.getsyspath(pathname)
def isfile(self, path):
path = self.getsyspath(path)
return os.path.isfile(path)
def ishidden(self, pathname):
return pathname.startswith('.')
def ishidden(self, path):
return path.startswith('.')
def listdir(self, path="./", wildcard=None, full=False, absolute=False, hidden=False, dirs_only=False, files_only=False):
try:
paths = os.listdir(self.getsyspath(path))
except (OSError, IOError), e:
raise FSError("LISTDIR_FAILED", path, details=e, msg="Unable to get directory listing: %(path)s - (%(details)s)")
raise OperationFailedError("LISTDIR_FAILED", path, details=e, msg="Unable to get directory listing: %(path)s - (%(details)s)")
return self._listdir_helper(path, paths, wildcard, full, absolute, hidden, dirs_only, files_only)
def mkdir(self, path, mode=0777, recursive=False):
sys_path = self.getsyspath(path)
if recursive:
......@@ -75,44 +63,37 @@ class OSFS(FS):
def remove(self, path):
sys_path = self.getsyspath(path)
try:
os.remove(sys_path)
except OSError, e:
raise FSError("FILE_DELETE_FAILED", path, details=e)
raise OperationFailedError("REMOVE_FAILED", path, details=e)
def removedir(self, path, recursive=False):
sys_path = self.getsyspath(path)
if recursive:
try:
os.rmdir(sys_path)
except OSError, e:
raise FSError("DIR_DELETE_FAILED", path, details=e)
raise OperationFailedError("REMOVEDIR_FAILED", path, details=e)
else:
try:
os.removedirs(sys_path)
except OSError, e:
raise FSError("DIR_DELETE_FAILED", path, details=e)
raise OperationFailedError("REMOVEDIR_FAILED", path, details=e)
def rename(self, src, dst):
path_src = self.getsyspath(src)
path_dst = self.getsyspath(dst)
try:
os.rename(path_src, path_dst)
except OSError, e:
raise FSError("RENAME_FAILED", src)
raise OperationFailedError("RENAME_FAILED", src)
def getinfo(self, path):
sys_path = self.getsyspath(path)
try:
......@@ -140,7 +121,6 @@ class OSFS(FS):
def getsize(self, path):
sys_path = self.getsyspath(path)
try:
......
fs/tests.py
......@@ -51,7 +51,6 @@ class TestHelpers(unittest.TestCase):
self.assertRaises(fs.PathError, fs.pathjoin, "a/b/../../../d")
def test_makerelative(self):
tests = [ ("/a/b", "a/b"),
("a/b", "a/b"),
("/", "") ]
......@@ -61,7 +60,6 @@ class TestHelpers(unittest.TestCase):
self.assertEqual(fs.makerelative(path), result)
def test_absolute(self):
tests = [ ("/a/b", "/a/b"),
("a/b", "/a/b"),
("/", "/") ]
......@@ -70,7 +68,6 @@ class TestHelpers(unittest.TestCase):
self.assertEqual(fs.makeabsolute(path), result)
def test_iteratepath(self):
tests = [ ("a/b", ["a", "b"]),
("", [] ),
("aaa/bbb/ccc", ["aaa", "bbb", "ccc"]),
......@@ -85,7 +82,6 @@ class TestHelpers(unittest.TestCase):
self.assertEqual(list(fs._iteratepath("a/b/c/d", 2)), ["a", "b", "c/d"])
def test_pathsplit(self):
tests = [ ("a/b", ("a", "b")),
("a/b/c", ("a/b", "c")),
("a", ("", "a")),
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment