Skip to content

P
pyfs
Commit bc30657b by willmcgugan@gmail.com
Options

Fixes for backslashes on Linux issue, see Issue #139

parent 2fbb136c
Showing with 133 additions and 110 deletions
fs/errors.py
......@@ -24,7 +24,7 @@ __all__ = ['FSError',
'NoMetaError',
'NoPathURLError',
'ResourceNotFoundError',
'ResourceInvalidError',
'ResourceInvalidError',
'DestinationExistsError',
'DirectoryNotEmptyError',
'ParentDirectoryMissingError',
......@@ -42,6 +42,10 @@ from fs.path import *
from fs.local_functools import wraps
class InvalidPathError(Exception):
pass
class FSError(Exception):
"""Base exception class for the FS module."""
default_message = "Unspecified error"
......@@ -81,7 +85,7 @@ class PathError(FSError):
def __init__(self,path="",**kwds):
self.path = path
super(PathError,self).__init__(**kwds)
class OperationFailedError(FSError):
"""Base exception class for errors associated with a specific operation."""
......@@ -184,6 +188,7 @@ class ResourceLockedError(ResourceError):
"""Exception raised when a resource can't be used because it is locked."""
default_message = "Resource is locked: %(path)s"
class NoMMapError(ResourceError):
"""Exception raise when getmmap fails to create a mmap"""
default_message = "Can't get mmap for %(path)s"
......
fs/osfs/__init__.py
......@@ -32,13 +32,15 @@ from fs.osfs.watch import OSFSWatchMixin
@convert_os_errors
def _os_stat(path):
"""Replacement for os.stat that raises FSError subclasses."""
"""Replacement for os.stat that raises FSError subclasses."""
return os.stat(path)
@convert_os_errors
def _os_mkdir(name, mode=0777):
"""Replacement for os.mkdir that raises FSError subclasses."""
return os.mkdir(name,mode)
return os.mkdir(name, mode)
@convert_os_errors
def _os_makedirs(name, mode=0777):
......@@ -64,7 +66,6 @@ def _os_makedirs(name, mode=0777):
if tail == os.curdir:
return
os.mkdir(name, mode)
class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
......@@ -74,7 +75,7 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
filesystem of the OS. Most of its methods simply defer to the matching
methods in the os and os.path modules.
"""
_meta = { 'thread_safe' : True,
'network' : False,
'virtual' : False,
......@@ -90,7 +91,7 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
"""
Creates an FS object that represents the OS Filesystem under a given root path
:param root_path: The root OS path
:param root_path: The root OS path
:param thread_synchronize: If True, this object will be thread-safe by use of a threading.Lock object
:param encoding: The encoding method for path strings
:param create: If True, then root_path will be created if it doesn't already exist
......@@ -114,7 +115,7 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
if root_path.startswith("\\\\"):
root_path = u"\\\\?\\UNC\\" + root_path[2:]
else:
root_path = u"\\\\?" + root_path
root_path = u"\\\\?" + root_path
# If it points at the root of a drive, it needs a trailing slash.
if len(root_path) == 6 and not root_path.endswith("\\"):
root_path = root_path + "\\"
......@@ -126,9 +127,9 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
pass
if not os.path.exists(root_path):
raise ResourceNotFoundError(root_path,msg="Root directory does not exist: %(path)s")
raise ResourceNotFoundError(root_path, msg="Root directory does not exist: %(path)s")
if not os.path.isdir(root_path):
raise ResourceInvalidError(root_path,msg="Root path is not a directory: %(path)s")
raise ResourceInvalidError(root_path, msg="Root path is not a directory: %(path)s")
self.root_path = root_path
self.dir_mode = dir_mode
......@@ -137,20 +138,20 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
def __repr__(self):
return "<OSFS: %r>" % self.root_path
def __unicode__(self):
return u"<OSFS: %s>" % self.root_path
def _decode_path(self, p):
if isinstance(p, unicode):
return p
return p.decode(self.encoding, 'replace')
return p
return p.decode(self.encoding, 'replace')
def getsyspath(self, path, allow_none=False):
path = relpath(normpath(path)).replace("/",os.sep)
path = relpath(normpath(path)).replace("/", os.sep)
path = os.path.join(self.root_path, path)
if not path.startswith(self.root_path):
raise PathError(path,msg="OSFS given path outside root: %(path)s")
raise PathError(path, msg="OSFS given path outside root: %(path)s")
path = self._decode_path(path)
return path
......@@ -159,11 +160,11 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
This basically the reverse of getsyspath(). If the path does not
refer to a location within this filesystem, ValueError is raised.
:param path: a system path
:returns: a path within this FS object
:rtype: string
"""
path = os.path.normpath(os.path.abspath(path))
path = self._decode_path(path)
......@@ -173,11 +174,11 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
if not prefix.endswith(os.path.sep):
prefix += os.path.sep
if not os.path.normcase(path).startswith(prefix):
raise ValueError("path not within this FS: %s (%s)" % (os.path.normcase(path),prefix))
raise ValueError("path not within this FS: %s (%s)" % (os.path.normcase(path), prefix))
return normpath(path[len(self.root_path):])
def getmeta(self, meta_name, default=NoDefaultMeta):
if meta_name == 'free_space':
if platform.system() == 'Windows':
try:
......@@ -204,11 +205,11 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
else:
stat = os.statvfs(self.root_path)
return stat.f_blocks * stat.f_bsize
return super(OSFS, self).getmeta(meta_name, default)
@convert_os_errors
def open(self, path, mode="r", **kwargs):
def open(self, path, mode="r", **kwargs):
mode = ''.join(c for c in mode if c in 'rwabt+')
sys_path = self.getsyspath(path)
try:
......@@ -221,25 +222,25 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
raise
@convert_os_errors
def setcontents(self, path, contents, chunk_size=64*1024):
return super(OSFS,self).setcontents(path, contents, chunk_size)
def setcontents(self, path, contents, chunk_size=64 * 1024):
return super(OSFS, self).setcontents(path, contents, chunk_size)
@convert_os_errors
def exists(self, path):
def exists(self, path):
return _exists(self.getsyspath(path))
@convert_os_errors
def isdir(self, path):
def isdir(self, path):
return _isdir(self.getsyspath(path))
@convert_os_errors
def isfile(self, path):
def isfile(self, path):
return _isfile(self.getsyspath(path))
@convert_os_errors
def listdir(self, path="./", wildcard=None, full=False, absolute=False, dirs_only=False, files_only=False):
_decode_path = self._decode_path
paths = [_decode_path(p) for p in os.listdir(self.getsyspath(path))]
_decode_path = self._decode_path
paths = [_decode_path(p) for p in os.listdir(self.getsyspath(path))]
return self._listdir_helper(path, paths, wildcard, full, absolute, dirs_only, files_only)
@convert_os_errors
......@@ -252,16 +253,16 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
_os_mkdir(sys_path, self.dir_mode)
except DestinationExistsError:
if self.isfile(path):
raise ResourceInvalidError(path,msg="Cannot create directory, there's already a file of that name: %(path)s")
raise ResourceInvalidError(path, msg="Cannot create directory, there's already a file of that name: %(path)s")
if not allow_recreate:
raise DestinationExistsError(path,msg="Can not create a directory that already exists (try allow_recreate=True): %(path)s")
raise DestinationExistsError(path, msg="Can not create a directory that already exists (try allow_recreate=True): %(path)s")
except ResourceNotFoundError:
raise ParentDirectoryMissingError(path)
@convert_os_errors
def remove(self, path):
sys_path = self.getsyspath(path)
try:
try:
os.remove(sys_path)
except OSError, e:
if e.errno == errno.EACCES and sys.platform == "win32":
......@@ -275,7 +276,7 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
raise
@convert_os_errors
def removedir(self, path, recursive=False, force=False):
def removedir(self, path, recursive=False, force=False):
sys_path = self.getsyspath(path)
if force:
for path2 in self.listdir(path, absolute=True, files_only=True):
......@@ -297,7 +298,7 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
if recursive:
try:
if dirname(path) not in ('', '/'):
self.removedir(dirname(path),recursive=True)
self.removedir(dirname(path), recursive=True)
except DirectoryNotEmptyError:
pass
......@@ -320,9 +321,9 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
if e.errno == errno.ENOENT:
if not os.path.exists(os.path.dirname(path_dst)):
raise ParentDirectoryMissingError(dst)
raise
def _stat(self,path):
raise
def _stat(self, path):
"""Stat the given path, normalising error codes."""
sys_path = self.getsyspath(path)
try:
......@@ -350,5 +351,3 @@ class OSFS(OSFSXAttrMixin, OSFSWatchMixin, FS):
@convert_os_errors
def getsize(self, path):
return self._stat(path).st_size
fs/path.py
from __future__ import unicode_literals
"""
fs.path
=======
......@@ -11,22 +13,23 @@ by forward slashes and with an optional leading slash).
"""
import re
import os
_requires_normalization = re.compile(r'/\.\.|\./|\.|//').search
_requires_normalization = re.compile(r'/\.\.|\./|\.|//|\\').search
def normpath(path):
"""Normalizes a path to be in the format expected by FS objects.
This function remove any leading or trailing slashes, collapses
duplicate slashes, replaces backward with forward slashes, and generally
tries very hard to return a new path string the canonical FS format.
duplicate slashes, and generally tries very hard to return a new path
in the canonical FS format.
If the path is invalid, ValueError will be raised.
:param path: path to normalize
:returns: a valid FS path
>>> normpath(r"foo\\bar\\baz")
'foo/bar/baz'
>>> normpath("/foo//bar/frob/../baz")
'/foo/bar/baz'
......@@ -40,15 +43,13 @@ def normpath(path):
if path in ('', '/'):
return path
path = path.replace('\\', '/')
# An early out if there is no need to normalize this path
if not _requires_normalization(path):
return path.rstrip('/')
components = []
append = components.append
special = ('..', '.', '').__contains__
special = ('..', '.', '').__contains__
try:
for component in path.split('/'):
if special(component):
......@@ -66,12 +67,27 @@ def normpath(path):
return '/'.join(components)
if os.sep != '/':
def ospath(path):
"""Replace path separators in an OS path if required"""
return path.replace(os.sep, '/')
else:
def ospath(path):
"""Replace path separators in an OS path if required"""
return path
def normospath(path):
"""Normalizes a path with os separators"""
return normpath(ospath)
def iteratepath(path, numsplits=None):
"""Iterate over the individual components of a path.
:param path: Path to iterate over
:numsplits: Maximum number of splits
"""
path = relpath(normpath(path))
if not path:
......@@ -84,39 +100,40 @@ def iteratepath(path, numsplits=None):
def recursepath(path, reverse=False):
"""Returns intermediate paths from the root to the given path
:param reverse: reverses the order of the paths
>>> recursepath('a/b/c')
['/', u'/a', u'/a/b', u'/a/b/c']
"""
"""
if path in ('', '/'):
return [u'/']
path = abspath(normpath(path)) + '/'
path = abspath(normpath(path)) + '/'
paths = [u'/']
find = path.find
append = paths.append
append = paths.append
pos = 1
len_path = len(path)
while pos < len_path:
pos = find('/', pos)
len_path = len(path)
while pos < len_path:
pos = find('/', pos)
append(path[:pos])
pos += 1
pos += 1
if reverse:
return paths[::-1]
return paths
return paths
def isabs(path):
"""Return True if path is an absolute path."""
return path.startswith('/')
def abspath(path):
"""Convert the given path to an absolute path.
......@@ -134,9 +151,9 @@ def relpath(path):
This is the inverse of abspath(), stripping a leading '/' from the
path if it is present.
:param path: Path to adjust
>>> relpath('/a/b')
'a/b'
......@@ -146,7 +163,7 @@ def relpath(path):
def pathjoin(*paths):
"""Joins any number of paths together, returning a new path string.
:param paths: Paths to join are given in positional arguments
>>> pathjoin('foo', 'bar', 'baz')
......@@ -160,10 +177,10 @@ def pathjoin(*paths):
"""
absolute = False
relpaths = []
relpaths = []
for p in paths:
if p:
if p[0] in '\\/':
if p[0] == '/':
del relpaths[:]
absolute = True
relpaths.append(p)
......@@ -173,24 +190,26 @@ def pathjoin(*paths):
path = abspath(path)
return path
def pathcombine(path1, path2):
"""Joins two paths together.
This is faster than `pathjoin`, but only works when the second path is relative,
and there are no backreferences in either path.
and there are no backreferences in either path.
>>> pathcombine("foo/bar", "baz")
'foo/bar/baz'
"""
'foo/bar/baz'
"""
return "%s/%s" % (path1.rstrip('/'), path2.lstrip('/'))
def join(*paths):
"""Joins any number of paths together, returning a new path string.
This is a simple alias for the ``pathjoin`` function, allowing it to be
used as ``fs.path.join`` in direct correspondence with ``os.path.join``.
:param paths: Paths to join are given in positional arguments
"""
return pathjoin(*paths)
......@@ -201,7 +220,7 @@ def pathsplit(path):
This function splits a path into a pair (head, tail) where 'tail' is the
last pathname component and 'head' is all preceding components.
:param path: Path to split
>>> pathsplit("foo/bar")
......@@ -209,7 +228,7 @@ def pathsplit(path):
>>> pathsplit("foo/bar/baz")
('foo/bar', 'baz')
>>> pathsplit("/foo/bar/baz")
('/foo/bar', 'baz')
......@@ -234,17 +253,17 @@ def split(path):
def splitext(path):
"""Splits the extension from the path, and returns the path (up to the last
'.' and the extension).
:param path: A path to split
>>> splitext('baz.txt')
('baz', 'txt')
>>> splitext('foo/bar/baz.txt')
('foo/bar/baz', 'txt')
"""
parent_path, pathname = pathsplit(path)
if '.' not in pathname:
return path, ''
......@@ -256,18 +275,18 @@ def splitext(path):
def isdotfile(path):
"""Detects if a path references a dot file, i.e. a resource who's name
starts with a '.'
:param path: Path to check
>>> isdotfile('.baz')
True
>>> isdotfile('foo/bar/baz')
True
>>> isdotfile('foo/bar.baz').
False
"""
return basename(path).startswith('.')
......@@ -277,15 +296,15 @@ def dirname(path):
This is always equivalent to the 'head' component of the value returned
by pathsplit(path).
:param path: A FS path
>>> dirname('foo/bar/baz')
'foo/bar'
>>> dirname('/foo/bar')
'/foo'
>>> dirname('/foo')
'/'
......@@ -298,15 +317,15 @@ def basename(path):
This is always equivalent to the 'tail' component of the value returned
by pathsplit(path).
:param path: A FS path
>>> basename('foo/bar/baz')
'baz'
>>> basename('foo/bar')
'bar'
>>> basename('foo/bar/')
''
......@@ -316,7 +335,7 @@ def basename(path):
def issamedir(path1, path2):
"""Return true if two paths reference a resource in the same directory.
:param path1: An FS path
:param path2: An FS path
......@@ -332,15 +351,15 @@ def issamedir(path1, path2):
def isbase(path1, path2):
p1 = forcedir(abspath(path1))
p2 = forcedir(abspath(path2))
return p1 == p2 or p1.startswith(p2)
return p1 == p2 or p1.startswith(p2)
def isprefix(path1, path2):
"""Return true is path1 is a prefix of path2.
:param path1: An FS path
:param path2: An FS path
>>> isprefix("foo/bar", "foo/bar/spam.txt")
True
>>> isprefix("foo/bar/", "foo/bar")
......@@ -365,7 +384,7 @@ def isprefix(path1, path2):
def forcedir(path):
"""Ensure the path ends with a trailing /
:param path: An FS path
>>> forcedir("foo/bar")
......@@ -602,12 +621,12 @@ class PathMap(object):
_wild_chars = frozenset('*?[]!{}')
def iswildcard(path):
"""Check if a path ends with a wildcard
>>> is_wildcard('foo/bar/baz.*')
True
>>> is_wildcard('foo/bar')
False
"""
assert path is not None
base_chars = frozenset(basename(path))
......
fs/tests/test_path.py
......@@ -14,7 +14,7 @@ class TestPathFunctions(unittest.TestCase):
"""Testcases for FS path functions."""
def test_normpath(self):
tests = [ ("\\a\\b\\c", "/a/b/c"),
tests = [ ("\\a\\b\\c", "\\a\\b\\c"),
(".", ""),
("./", ""),
("", ""),
......@@ -22,7 +22,7 @@ class TestPathFunctions(unittest.TestCase):
("a/b/c", "a/b/c"),
("a/b/../c/", "a/c"),
("/","/"),
(u"a/\N{GREEK SMALL LETTER BETA}\\c",u"a/\N{GREEK SMALL LETTER BETA}/c"),
(u"a/\N{GREEK SMALL LETTER BETA}/c",u"a/\N{GREEK SMALL LETTER BETA}/c"),
]
for path, result in tests:
self.assertEqual(normpath(path), result)
......@@ -38,7 +38,7 @@ class TestPathFunctions(unittest.TestCase):
("a/b/c", "../d", "c", "a/b/d/c"),
("a/b/c", "../d", "/a", "/a"),
("aaa", "bbb/ccc", "aaa/bbb/ccc"),
("aaa", "bbb\ccc", "aaa/bbb/ccc"),
("aaa", "bbb\\ccc", "aaa/bbb\\ccc"),
("aaa", "bbb", "ccc", "/aaa", "eee", "/aaa/eee"),
("a/b", "./d", "e", "a/b/d/e"),
("/", "/", "/"),
......@@ -104,7 +104,7 @@ class TestPathFunctions(unittest.TestCase):
self.assertEquals(recursepath("/hello/world/",reverse=True),["/hello/world","/hello","/"])
self.assertEquals(recursepath("hello",reverse=True),["/hello","/"])
self.assertEquals(recursepath("",reverse=True),["/"])
def test_isdotfile(self):
for path in ['.foo',
'.svn',
......@@ -112,14 +112,14 @@ class TestPathFunctions(unittest.TestCase):
'foo/bar/.svn',
'/foo/.bar']:
self.assert_(isdotfile(path))
for path in ['asfoo',
'df.svn',
'foo/er.svn',
'foo/bar/test.txt',
'/foo/bar']:
self.assertFalse(isdotfile(path))
def test_dirname(self):
tests = [('foo', ''),
('foo/bar', 'foo'),
......@@ -129,7 +129,7 @@ class TestPathFunctions(unittest.TestCase):
('/', '/')]
for path, test_dirname in tests:
self.assertEqual(dirname(path), test_dirname)
def test_basename(self):
tests = [('foo', 'foo'),
('foo/bar', 'bar'),
......
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