move localpath implementation to a single file, simplify unix/posix difference and fix a bit
--HG-- branch : trunk
This commit is contained in:
parent
079a2327ec
commit
561fdca3a2
|
@ -1,6 +1,6 @@
|
|||
|
||||
|
||||
py.test --dist=each $* \
|
||||
--tx 'popen//python=python2.6' \
|
||||
--tx 'ssh=noco//python=/usr/local/bin/python2.4//chdir=/tmp/pytest-python2.4' \
|
||||
--tx 'socket=192.168.1.106:8888'
|
||||
#--tx 'popen//python=python2.6' \
|
||||
#--tx 'ssh=noco//python=/usr/local/bin/python2.4//chdir=/tmp/pytest-python2.4' \
|
||||
|
|
|
@ -1,3 +1,8 @@
|
|||
Changes between 1.0.x and 'trunk'
|
||||
=====================================
|
||||
|
||||
* simplified internal localpath implementation
|
||||
|
||||
Changes between 1.0.0 and 1.0.1
|
||||
=====================================
|
||||
|
||||
|
|
|
@ -106,7 +106,7 @@ initpkg(__name__,
|
|||
'path.__doc__' : ('./path/__init__.py', '__doc__'),
|
||||
'path.svnwc' : ('./path/svn/wccommand.py', 'SvnWCCommandPath'),
|
||||
'path.svnurl' : ('./path/svn/urlcommand.py', 'SvnCommandPath'),
|
||||
'path.local' : ('./path/local/local.py', 'LocalPath'),
|
||||
'path.local' : ('./path/local.py', 'LocalPath'),
|
||||
'path.SvnAuth' : ('./path/svn/svncommon.py', 'SvnAuth'),
|
||||
|
||||
# some nice slightly magic APIs
|
||||
|
|
|
@ -1,9 +1,5 @@
|
|||
"""
|
||||
specialized local path implementation.
|
||||
|
||||
This Path implementation offers some methods like chmod(), owner()
|
||||
and so on that may only make sense on unix.
|
||||
|
||||
local path implementation.
|
||||
"""
|
||||
from __future__ import generators
|
||||
import sys, os, stat, re, atexit
|
||||
|
@ -12,17 +8,93 @@ from py.__.path import common
|
|||
|
||||
iswin32 = sys.platform == "win32"
|
||||
|
||||
if iswin32:
|
||||
from py.__.path.local.win import WinMixin as PlatformMixin
|
||||
else:
|
||||
from py.__.path.local.posix import PosixMixin as PlatformMixin
|
||||
class Stat(object):
|
||||
pformat = "%s = property(lambda x: getattr(x._osstatresult, 'st_%s'))"
|
||||
|
||||
class LocalPath(common.FSPathBase, PlatformMixin):
|
||||
""" Local path implementation offering access/modification
|
||||
methods similar to os.path.
|
||||
for name in ('atime blksize blocks ctime dev gid '
|
||||
'ino mode mtime nlink rdev size uid'.split()):
|
||||
code = pformat.replace("%s", name)
|
||||
exec code
|
||||
|
||||
def __init__(self, path, osstatresult):
|
||||
self.path = path
|
||||
self._osstatresult = osstatresult
|
||||
|
||||
def owner(self):
|
||||
if iswin32:
|
||||
raise NotImplementedError("XXX win32")
|
||||
import pwd
|
||||
entry = self.path._callex(pwd.getpwuid, self.uid)
|
||||
return entry[0]
|
||||
owner = property(owner, None, None, "owner of path")
|
||||
|
||||
def group(self):
|
||||
""" return group name of file. """
|
||||
if iswin32:
|
||||
raise NotImplementedError("XXX win32")
|
||||
import grp
|
||||
entry = self.path._callex(grp.getgrgid, self.gid)
|
||||
return entry[0]
|
||||
group = property(group)
|
||||
|
||||
class PosixPath(common.FSPathBase):
|
||||
def chown(self, user, group, rec=0):
|
||||
""" change ownership to the given user and group.
|
||||
user and group may be specified by a number or
|
||||
by a name. if rec is True change ownership
|
||||
recursively.
|
||||
"""
|
||||
uid = getuserid(user)
|
||||
gid = getgroupid(group)
|
||||
if rec:
|
||||
for x in self.visit(rec=lambda x: x.check(link=0)):
|
||||
if x.check(link=0):
|
||||
self._callex(os.chown, str(x), uid, gid)
|
||||
self._callex(os.chown, str(self), uid, gid)
|
||||
|
||||
def readlink(self):
|
||||
""" return value of a symbolic link. """
|
||||
return self._callex(os.readlink, self.strpath)
|
||||
|
||||
def mklinkto(self, oldname):
|
||||
""" posix style hard link to another name. """
|
||||
self._callex(os.link, str(oldname), str(self))
|
||||
|
||||
def mksymlinkto(self, value, absolute=1):
|
||||
""" create a symbolic link with the given value (pointing to another name). """
|
||||
if absolute:
|
||||
self._callex(os.symlink, str(value), self.strpath)
|
||||
else:
|
||||
base = self.common(value)
|
||||
# with posix local paths '/' is always a common base
|
||||
relsource = self.__class__(value).relto(base)
|
||||
reldest = self.relto(base)
|
||||
n = reldest.count(self.sep)
|
||||
target = self.sep.join(('..', )*n + (relsource, ))
|
||||
self._callex(os.symlink, target, self.strpath)
|
||||
|
||||
def samefile(self, other):
|
||||
""" return True if other refers to the same stat object as self. """
|
||||
return py.std.os.path.samefile(str(self), str(other))
|
||||
|
||||
def getuserid(user):
|
||||
import pwd
|
||||
if not isinstance(user, int):
|
||||
user = pwd.getpwnam(user)[2]
|
||||
return user
|
||||
|
||||
def getgroupid(group):
|
||||
import grp
|
||||
if not isinstance(group, int):
|
||||
group = grp.getgrnam(group)[2]
|
||||
return group
|
||||
|
||||
FSBase = not iswin32 and PosixPath or common.FSPathBase
|
||||
|
||||
class LocalPath(FSBase):
|
||||
""" object oriented interface to os.path and other local filesystem
|
||||
related information.
|
||||
"""
|
||||
_path_cache = {}
|
||||
|
||||
sep = os.sep
|
||||
class Checkers(common.FSCheckers):
|
||||
def _stat(self):
|
||||
|
@ -76,6 +148,21 @@ class LocalPath(common.FSPathBase, PlatformMixin):
|
|||
def __hash__(self):
|
||||
return hash(self.strpath)
|
||||
|
||||
def remove(self, rec=1):
|
||||
""" remove a file or directory (or a directory tree if rec=1). """
|
||||
if self.check(dir=1, link=0):
|
||||
if rec:
|
||||
# force remove of readonly files on windows
|
||||
if iswin32:
|
||||
self.chmod(0700, rec=1)
|
||||
self._callex(py.std.shutil.rmtree, self.strpath)
|
||||
else:
|
||||
self._callex(os.rmdir, self.strpath)
|
||||
else:
|
||||
if iswin32:
|
||||
self.chmod(0700)
|
||||
self._callex(os.remove, self.strpath)
|
||||
|
||||
def computehash(self, hashtype="md5", chunksize=524288):
|
||||
""" return hexdigest of hashvalue for this file. """
|
||||
hash = self._gethashinstance(hashtype)
|
||||
|
@ -311,14 +398,12 @@ class LocalPath(common.FSPathBase, PlatformMixin):
|
|||
|
||||
def stat(self):
|
||||
""" Return an os.stat() tuple. """
|
||||
stat = self._callex(os.stat, self.strpath)
|
||||
return self._makestat(stat)
|
||||
return Stat(self, self._callex(os.stat, self.strpath))
|
||||
|
||||
def lstat(self):
|
||||
""" Return an os.lstat() tuple. """
|
||||
return self._makestat(self._callex(os.lstat, self.strpath))
|
||||
return Stat(self, self._callex(os.lstat, self.strpath))
|
||||
|
||||
# xlocal implementation
|
||||
def setmtime(self, mtime=None):
|
||||
""" set modification time for the given path. if 'mtime' is None
|
||||
(the default) then the file's mtime is set to current time.
|
||||
|
@ -379,6 +464,18 @@ class LocalPath(common.FSPathBase, PlatformMixin):
|
|||
#print "prepending to sys.path", s
|
||||
sys.path.insert(0, s)
|
||||
|
||||
def chmod(self, mode, rec=0):
|
||||
""" change permissions to the given mode. If mode is an
|
||||
integer it directly encodes the os-specific modes.
|
||||
if rec is True perform recursively.
|
||||
"""
|
||||
if not isinstance(mode, int):
|
||||
raise TypeError("mode %r must be an integer" % (mode,))
|
||||
if rec:
|
||||
for x in self.visit(rec=rec):
|
||||
self._callex(os.chmod, str(x), mode)
|
||||
self._callex(os.chmod, str(self), mode)
|
||||
|
||||
def pyimport(self, modname=None, ensuresyspath=True):
|
||||
""" return path as an imported python module.
|
||||
if modname is None, look for the containing package
|
||||
|
@ -438,7 +535,7 @@ class LocalPath(common.FSPathBase, PlatformMixin):
|
|||
|
||||
def _getpycodeobj(self):
|
||||
""" read the path and compile it to a code object. """
|
||||
dotpy = self.check(ext='.py')
|
||||
dotpy = self.ext == ".py"
|
||||
if dotpy:
|
||||
my_magic = py.std.imp.get_magic()
|
||||
my_timestamp = int(self.mtime())
|
|
@ -1 +0,0 @@
|
|||
#
|
|
@ -1,25 +0,0 @@
|
|||
import py
|
||||
|
||||
class Stat(object):
|
||||
def __init__(self, path, osstatresult):
|
||||
self.path = path
|
||||
self._osstatresult = osstatresult
|
||||
|
||||
for name in ('atime blksize blocks ctime dev gid '
|
||||
'ino mode mtime nlink rdev size uid'.split()):
|
||||
|
||||
code = """if 1:
|
||||
def fget(self):
|
||||
return getattr(self._osstatresult, "st_%(name)s", None)
|
||||
%(name)s = property(fget)
|
||||
def fget_deprecated(self):
|
||||
py.std.warnings.warn("statresult.st_%(name)s is deprecated, use "
|
||||
"statresult.%(name)s instead.",
|
||||
DeprecationWarning, stacklevel=2)
|
||||
return getattr(self._osstatresult, "st_%(name)s", None)
|
||||
st_%(name)s = property(fget_deprecated)
|
||||
""" % locals()
|
||||
exec code
|
||||
del fget
|
||||
del fget_deprecated
|
||||
|
|
@ -1,129 +0,0 @@
|
|||
"""
|
||||
module to access local filesystem pathes
|
||||
(mostly filename manipulations but also file operations)
|
||||
"""
|
||||
import os, sys, stat
|
||||
|
||||
import py
|
||||
#__________________________________________________________
|
||||
#
|
||||
# Local Path Posix Mixin
|
||||
#__________________________________________________________
|
||||
|
||||
from py.__.path.local.common import Stat
|
||||
|
||||
class PosixStat(Stat):
|
||||
def owner(self):
|
||||
entry = self.path._callex(py.std.pwd.getpwuid, self.uid)
|
||||
return entry[0]
|
||||
owner = property(owner, None, None, "owner of path")
|
||||
|
||||
def group(self):
|
||||
""" return group name of file. """
|
||||
entry = self.path._callex(py.std.grp.getgrgid, self.gid)
|
||||
return entry[0]
|
||||
group = property(group)
|
||||
|
||||
class PosixMixin(object):
|
||||
def _makestat(self, statresult):
|
||||
return PosixStat(self, statresult)
|
||||
|
||||
def _deprecated(self, name):
|
||||
py.std.warnings.warn("'path.%s()' is deprecated, use "
|
||||
"'path.stat().%s' instead." % (name,name),
|
||||
DeprecationWarning, stacklevel=3)
|
||||
|
||||
# an instance needs to be a local path instance
|
||||
def owner(self):
|
||||
""" return owner name of file. """
|
||||
self._deprecated('owner')
|
||||
return self.stat().owner
|
||||
|
||||
def group(self):
|
||||
""" return group name of file. """
|
||||
self._deprecated('group')
|
||||
return self.stat().group
|
||||
|
||||
def mode(self):
|
||||
""" return permission mode of the path object """
|
||||
self._deprecated('mode')
|
||||
return self.stat().mode
|
||||
|
||||
def chmod(self, mode, rec=0):
|
||||
""" change permissions to the given mode. If mode is an
|
||||
integer it directly encodes the os-specific modes.
|
||||
if rec is True perform recursively.
|
||||
|
||||
(xxx if mode is a string then it specifies access rights
|
||||
in '/bin/chmod' style, e.g. a+r).
|
||||
"""
|
||||
if not isinstance(mode, int):
|
||||
raise TypeError("mode %r must be an integer" % (mode,))
|
||||
if rec:
|
||||
for x in self.visit(rec=rec):
|
||||
self._callex(os.chmod, str(x), mode)
|
||||
self._callex(os.chmod, str(self), mode)
|
||||
|
||||
def chown(self, user, group, rec=0):
|
||||
""" change ownership to the given user and group.
|
||||
user and group may be specified by a number or
|
||||
by a name. if rec is True change ownership
|
||||
recursively.
|
||||
"""
|
||||
uid = getuserid(user)
|
||||
gid = getgroupid(group)
|
||||
if rec:
|
||||
for x in self.visit(rec=lambda x: x.check(link=0)):
|
||||
if x.check(link=0):
|
||||
self._callex(os.chown, str(x), uid, gid)
|
||||
self._callex(os.chown, str(self), uid, gid)
|
||||
|
||||
def readlink(self):
|
||||
""" return value of a symbolic link. """
|
||||
return self._callex(os.readlink, self.strpath)
|
||||
|
||||
def mklinkto(self, oldname):
|
||||
""" posix style hard link to another name. """
|
||||
self._callex(os.link, str(oldname), str(self))
|
||||
|
||||
def mksymlinkto(self, value, absolute=1):
|
||||
""" create a symbolic link with the given value (pointing to another name). """
|
||||
if absolute:
|
||||
self._callex(os.symlink, str(value), self.strpath)
|
||||
else:
|
||||
base = self.common(value)
|
||||
# with posix local paths '/' is always a common base
|
||||
relsource = self.__class__(value).relto(base)
|
||||
reldest = self.relto(base)
|
||||
n = reldest.count(self.sep)
|
||||
target = self.sep.join(('..', )*n + (relsource, ))
|
||||
self._callex(os.symlink, target, self.strpath)
|
||||
|
||||
|
||||
def remove(self, rec=1):
|
||||
""" remove a file or directory (or a directory tree if rec=1). """
|
||||
if self.check(dir=1, link=0):
|
||||
if rec:
|
||||
self._callex(py.std.shutil.rmtree, self.strpath)
|
||||
else:
|
||||
self._callex(os.rmdir, self.strpath)
|
||||
else:
|
||||
self._callex(os.remove, self.strpath)
|
||||
|
||||
def samefile(self, other):
|
||||
""" return True if other refers to the same stat object as self. """
|
||||
return py.std.os.path.samefile(str(self), str(other))
|
||||
|
||||
def getuserid(user):
|
||||
import pwd
|
||||
if isinstance(user, int):
|
||||
return user
|
||||
entry = pwd.getpwnam(user)
|
||||
return entry[2]
|
||||
|
||||
def getgroupid(group):
|
||||
import grp
|
||||
if isinstance(group, int):
|
||||
return group
|
||||
entry = grp.getgrnam(group)
|
||||
return entry[2]
|
|
@ -1 +0,0 @@
|
|||
#
|
|
@ -1,190 +0,0 @@
|
|||
import py
|
||||
|
||||
class TestPOSIXLocalPath:
|
||||
disabled = py.std.sys.platform == 'win32'
|
||||
|
||||
def setup_class(cls):
|
||||
cls.root = py.test.ensuretemp(cls.__name__)
|
||||
|
||||
def setup_method(self, method):
|
||||
name = method.im_func.func_name
|
||||
self.tmpdir = self.root.ensure(name, dir=1)
|
||||
|
||||
def test_samefile(self):
|
||||
assert self.tmpdir.samefile(self.tmpdir)
|
||||
p = self.tmpdir.ensure("hello")
|
||||
assert p.samefile(p)
|
||||
|
||||
def test_hardlink(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("Hello")
|
||||
nlink = filepath.stat().st_nlink
|
||||
linkpath.mklinkto(filepath)
|
||||
assert filepath.stat().st_nlink == nlink + 1
|
||||
|
||||
def test_symlink_are_identical(self):
|
||||
tmpdir = self.tmpdir
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("Hello")
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(filepath)
|
||||
assert linkpath.readlink() == str(filepath)
|
||||
|
||||
def test_symlink_isfile(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("")
|
||||
linkpath.mksymlinkto(filepath)
|
||||
assert linkpath.check(file=1)
|
||||
assert not linkpath.check(link=0, file=1)
|
||||
|
||||
def test_symlink_relative(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("Hello")
|
||||
linkpath.mksymlinkto(filepath, absolute=False)
|
||||
assert linkpath.readlink() == "file"
|
||||
assert filepath.read() == linkpath.read()
|
||||
|
||||
def test_symlink_not_existing(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('testnotexisting')
|
||||
assert not linkpath.check(link=1)
|
||||
assert linkpath.check(link=0)
|
||||
|
||||
def test_relto_with_root(self):
|
||||
y = self.root.join('x').relto(py.path.local('/'))
|
||||
assert y[0] == str(self.root)[1]
|
||||
|
||||
def test_visit_recursive_symlink(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(tmpdir)
|
||||
visitor = tmpdir.visit(None, lambda x: x.check(link=0))
|
||||
assert list(visitor) == [linkpath]
|
||||
|
||||
def test_symlink_isdir(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(tmpdir)
|
||||
assert linkpath.check(dir=1)
|
||||
assert not linkpath.check(link=0, dir=1)
|
||||
|
||||
def test_symlink_remove(self):
|
||||
tmpdir = self.tmpdir.realpath()
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(linkpath) # point to itself
|
||||
assert linkpath.check(link=1)
|
||||
linkpath.remove()
|
||||
assert not linkpath.check()
|
||||
|
||||
def test_realpath_file(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("")
|
||||
linkpath.mksymlinkto(filepath)
|
||||
realpath = linkpath.realpath()
|
||||
assert realpath.basename == 'file'
|
||||
|
||||
def test_owner(self):
|
||||
from pwd import getpwuid
|
||||
from grp import getgrgid
|
||||
stat = self.root.stat()
|
||||
assert stat.path == self.root
|
||||
|
||||
uid = stat.st_uid
|
||||
gid = stat.st_gid
|
||||
owner = getpwuid(uid)[0]
|
||||
group = getgrgid(gid)[0]
|
||||
|
||||
assert uid == stat.uid
|
||||
assert owner == stat.owner
|
||||
assert gid == stat.gid
|
||||
assert group == stat.group
|
||||
|
||||
def test_atime(self):
|
||||
import time
|
||||
path = self.root.ensure('samplefile')
|
||||
now = time.time()
|
||||
atime1 = path.atime()
|
||||
# we could wait here but timer resolution is very
|
||||
# system dependent
|
||||
path.read()
|
||||
atime2 = path.atime()
|
||||
duration = time.time() - now
|
||||
assert (atime2-atime1) <= duration
|
||||
|
||||
def test_commondir(self):
|
||||
# XXX This is here in local until we find a way to implement this
|
||||
# using the subversion command line api.
|
||||
p1 = self.root.join('something')
|
||||
p2 = self.root.join('otherthing')
|
||||
assert p1.common(p2) == self.root
|
||||
assert p2.common(p1) == self.root
|
||||
|
||||
def test_commondir_nocommon(self):
|
||||
# XXX This is here in local until we find a way to implement this
|
||||
# using the subversion command line api.
|
||||
p1 = self.root.join('something')
|
||||
p2 = py.path.local(self.root.sep+'blabla')
|
||||
assert p1.common(p2) == '/'
|
||||
|
||||
def test_join_to_root(self):
|
||||
root = self.root.parts()[0]
|
||||
assert len(str(root)) == 1
|
||||
assert str(root.join('a')) == '/a'
|
||||
|
||||
def test_join_root_to_root_with_no_abs(self):
|
||||
nroot = self.root.join('/')
|
||||
assert str(self.root) == str(nroot)
|
||||
assert self.root == nroot
|
||||
|
||||
def test_chmod_simple_int(self):
|
||||
print "self.root is", self.root
|
||||
mode = self.root.mode()
|
||||
self.root.chmod(mode/2)
|
||||
try:
|
||||
assert self.root.mode() != mode
|
||||
finally:
|
||||
self.root.chmod(mode)
|
||||
assert self.root.mode() == mode
|
||||
|
||||
def test_chmod_rec_int(self):
|
||||
# XXX fragile test
|
||||
print "self.root is", self.root
|
||||
recfilter = lambda x: x.check(dotfile=0, link=0)
|
||||
oldmodes = {}
|
||||
for x in self.root.visit(rec=recfilter):
|
||||
oldmodes[x] = x.mode()
|
||||
self.root.chmod(0772, rec=recfilter)
|
||||
try:
|
||||
for x in self.root.visit(rec=recfilter):
|
||||
assert x.mode() & 0777 == 0772
|
||||
finally:
|
||||
for x,y in oldmodes.items():
|
||||
x.chmod(y)
|
||||
|
||||
def test_chown_identity(self):
|
||||
owner = self.root.owner()
|
||||
group = self.root.group()
|
||||
self.root.chown(owner, group)
|
||||
|
||||
def test_chown_dangling_link(self):
|
||||
owner = self.root.owner()
|
||||
group = self.root.group()
|
||||
x = self.root.join('hello')
|
||||
x.mksymlinkto('qlwkejqwlek')
|
||||
try:
|
||||
self.root.chown(owner, group, rec=1)
|
||||
finally:
|
||||
x.remove(rec=0)
|
||||
|
||||
def test_chown_identity_rec_mayfail(self):
|
||||
owner = self.root.owner()
|
||||
group = self.root.group()
|
||||
self.root.chown(owner, group)
|
|
@ -1,56 +0,0 @@
|
|||
import py
|
||||
|
||||
class TestWINLocalPath:
|
||||
#root = local(TestLocalPath.root)
|
||||
disabled = py.std.sys.platform != 'win32'
|
||||
|
||||
def setup_class(cls):
|
||||
cls.root = py.test.ensuretemp(cls.__name__)
|
||||
|
||||
def setup_method(self, method):
|
||||
name = method.im_func.func_name
|
||||
self.tmpdir = self.root.ensure(name, dir=1)
|
||||
|
||||
def test_chmod_simple_int(self):
|
||||
print "self.root is", self.root
|
||||
mode = self.root.stat().st_mode
|
||||
# Ensure that we actually change the mode to something different.
|
||||
self.root.chmod(mode == 0 and 1 or 0)
|
||||
try:
|
||||
print self.root.stat().st_mode
|
||||
print mode
|
||||
assert self.root.stat().st_mode != mode
|
||||
finally:
|
||||
self.root.chmod(mode)
|
||||
assert self.root.stat().st_mode == mode
|
||||
|
||||
def test_path_comparison_lowercase_mixed(self):
|
||||
t1 = self.root.join("a_path")
|
||||
t2 = self.root.join("A_path")
|
||||
assert t1 == t1
|
||||
assert t1 == t2
|
||||
|
||||
def test_relto_with_mixed_case(self):
|
||||
t1 = self.root.join("a_path", "fiLe")
|
||||
t2 = self.root.join("A_path")
|
||||
assert t1.relto(t2) == "fiLe"
|
||||
|
||||
def test_allow_unix_style_paths(self):
|
||||
t1 = self.root.join('a_path')
|
||||
assert t1 == str(self.root) + '\\a_path'
|
||||
t1 = self.root.join('a_path/')
|
||||
assert t1 == str(self.root) + '\\a_path'
|
||||
t1 = self.root.join('dir/a_path')
|
||||
assert t1 == str(self.root) + '\\dir\\a_path'
|
||||
|
||||
def test_sysfind_in_currentdir(self):
|
||||
cmd = py.path.local.sysfind('cmd')
|
||||
root = cmd.new(dirname='', basename='') # c:\ in most installations
|
||||
|
||||
old = root.chdir()
|
||||
try:
|
||||
x = py.path.local.sysfind(cmd.relto(root))
|
||||
assert x.check(file=1)
|
||||
finally:
|
||||
old.chdir()
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
"""
|
||||
module for win-specific local path stuff
|
||||
|
||||
(implementor needed :-)
|
||||
"""
|
||||
|
||||
import os
|
||||
import py
|
||||
from py.__.path.local.common import Stat
|
||||
|
||||
class WinMixin:
|
||||
def _makestat(self, statresult):
|
||||
return Stat(self, statresult)
|
||||
|
||||
def chmod(self, mode, rec=0):
|
||||
""" change permissions to the given mode. If mode is an
|
||||
integer it directly encodes the os-specific modes.
|
||||
if rec is True perform recursively.
|
||||
|
||||
(xxx if mode is a string then it specifies access rights
|
||||
in '/bin/chmod' style, e.g. a+r).
|
||||
"""
|
||||
if not isinstance(mode, int):
|
||||
raise TypeError("mode %r must be an integer" % (mode,))
|
||||
if rec:
|
||||
for x in self.visit(rec=rec):
|
||||
self._callex(os.chmod, str(x), mode)
|
||||
self._callex(os.chmod, str(self), mode)
|
||||
|
||||
def remove(self, rec=1):
|
||||
""" remove a file or directory (or a directory tree if rec=1). """
|
||||
if self.check(dir=1, link=0):
|
||||
if rec:
|
||||
# force remove of readonly files on windows
|
||||
self.chmod(0700, rec=1)
|
||||
self._callex(py.std.shutil.rmtree, self.strpath)
|
||||
else:
|
||||
self._callex(os.rmdir, self.strpath)
|
||||
else:
|
||||
self.chmod(0700)
|
||||
self._callex(os.remove, self.strpath)
|
|
@ -94,8 +94,6 @@ class CommonFSTests(common.CommonPathTests):
|
|||
assert not self.root.join("sampledir").check(file=1)
|
||||
assert self.root.join("sampledir").check(file=0)
|
||||
|
||||
#def test_fnmatch_dir(self):
|
||||
|
||||
def test_non_existent(self):
|
||||
assert self.root.join("sampledir.nothere").check(dir=0)
|
||||
assert self.root.join("sampledir.nothere").check(file=0)
|
||||
|
@ -205,8 +203,12 @@ class CommonFSTests(common.CommonPathTests):
|
|||
p = self.root.join('samplefile')
|
||||
newp = p.dirpath('moved_samplefile')
|
||||
p.move(newp)
|
||||
assert newp.check(file=1)
|
||||
assert not p.check()
|
||||
try:
|
||||
assert newp.check(file=1)
|
||||
assert not p.check()
|
||||
finally:
|
||||
newp.move(p)
|
||||
assert p.check()
|
||||
|
||||
def test_move_directory(self):
|
||||
source = self.root.join('sampledir')
|
||||
|
|
|
@ -12,8 +12,10 @@ class LocalSetup:
|
|||
def setup_method(self, method):
|
||||
self.tmpdir = self.root.mkdir(method.__name__)
|
||||
|
||||
class TestLocalPath(LocalSetup, CommonFSTests):
|
||||
def teardown_method(self, method):
|
||||
assert self.root.join("samplefile").check()
|
||||
|
||||
class TestLocalPath(LocalSetup, CommonFSTests):
|
||||
def test_join_normpath(self):
|
||||
assert self.tmpdir.join(".") == self.tmpdir
|
||||
p = self.tmpdir.join("../%s" % self.tmpdir.basename)
|
||||
|
@ -347,3 +349,248 @@ def test_homedir():
|
|||
homedir = py.path.local._gethomedir()
|
||||
assert homedir.check(dir=1)
|
||||
|
||||
class TestWINLocalPath:
|
||||
#root = local(TestLocalPath.root)
|
||||
disabled = py.std.sys.platform != 'win32'
|
||||
|
||||
def setup_class(cls):
|
||||
cls.root = py.test.ensuretemp(cls.__name__)
|
||||
|
||||
def setup_method(self, method):
|
||||
name = method.im_func.func_name
|
||||
self.tmpdir = self.root.ensure(name, dir=1)
|
||||
|
||||
def test_owner_group_not_implemented(self):
|
||||
py.test.raises(NotImplementedError, "self.root.stat().owner")
|
||||
py.test.raises(NotImplementedError, "self.root.stat().group")
|
||||
|
||||
def test_chmod_simple_int(self):
|
||||
print "self.root is", self.root
|
||||
mode = self.root.stat().mode
|
||||
# Ensure that we actually change the mode to something different.
|
||||
self.root.chmod(mode == 0 and 1 or 0)
|
||||
try:
|
||||
print self.root.stat().mode
|
||||
print mode
|
||||
assert self.root.stat().mode != mode
|
||||
finally:
|
||||
self.root.chmod(mode)
|
||||
assert self.root.stat().mode == mode
|
||||
|
||||
def test_path_comparison_lowercase_mixed(self):
|
||||
t1 = self.root.join("a_path")
|
||||
t2 = self.root.join("A_path")
|
||||
assert t1 == t1
|
||||
assert t1 == t2
|
||||
|
||||
def test_relto_with_mixed_case(self):
|
||||
t1 = self.root.join("a_path", "fiLe")
|
||||
t2 = self.root.join("A_path")
|
||||
assert t1.relto(t2) == "fiLe"
|
||||
|
||||
def test_allow_unix_style_paths(self):
|
||||
t1 = self.root.join('a_path')
|
||||
assert t1 == str(self.root) + '\\a_path'
|
||||
t1 = self.root.join('a_path/')
|
||||
assert t1 == str(self.root) + '\\a_path'
|
||||
t1 = self.root.join('dir/a_path')
|
||||
assert t1 == str(self.root) + '\\dir\\a_path'
|
||||
|
||||
def test_sysfind_in_currentdir(self):
|
||||
cmd = py.path.local.sysfind('cmd')
|
||||
root = cmd.new(dirname='', basename='') # c:\ in most installations
|
||||
old = root.chdir()
|
||||
try:
|
||||
x = py.path.local.sysfind(cmd.relto(root))
|
||||
assert x.check(file=1)
|
||||
finally:
|
||||
old.chdir()
|
||||
|
||||
class TestPOSIXLocalPath:
|
||||
disabled = py.std.sys.platform == 'win32'
|
||||
|
||||
def setup_class(cls):
|
||||
cls.root = py.test.ensuretemp(cls.__name__)
|
||||
|
||||
def setup_method(self, method):
|
||||
name = method.im_func.func_name
|
||||
self.tmpdir = self.root.ensure(name, dir=1)
|
||||
|
||||
def test_samefile(self):
|
||||
assert self.tmpdir.samefile(self.tmpdir)
|
||||
p = self.tmpdir.ensure("hello")
|
||||
assert p.samefile(p)
|
||||
|
||||
def test_hardlink(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("Hello")
|
||||
nlink = filepath.stat().nlink
|
||||
linkpath.mklinkto(filepath)
|
||||
assert filepath.stat().nlink == nlink + 1
|
||||
|
||||
def test_symlink_are_identical(self):
|
||||
tmpdir = self.tmpdir
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("Hello")
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(filepath)
|
||||
assert linkpath.readlink() == str(filepath)
|
||||
|
||||
def test_symlink_isfile(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("")
|
||||
linkpath.mksymlinkto(filepath)
|
||||
assert linkpath.check(file=1)
|
||||
assert not linkpath.check(link=0, file=1)
|
||||
|
||||
def test_symlink_relative(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("Hello")
|
||||
linkpath.mksymlinkto(filepath, absolute=False)
|
||||
assert linkpath.readlink() == "file"
|
||||
assert filepath.read() == linkpath.read()
|
||||
|
||||
def test_symlink_not_existing(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('testnotexisting')
|
||||
assert not linkpath.check(link=1)
|
||||
assert linkpath.check(link=0)
|
||||
|
||||
def test_relto_with_root(self):
|
||||
y = self.root.join('x').relto(py.path.local('/'))
|
||||
assert y[0] == str(self.root)[1]
|
||||
|
||||
def test_visit_recursive_symlink(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(tmpdir)
|
||||
visitor = tmpdir.visit(None, lambda x: x.check(link=0))
|
||||
assert list(visitor) == [linkpath]
|
||||
|
||||
def test_symlink_isdir(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(tmpdir)
|
||||
assert linkpath.check(dir=1)
|
||||
assert not linkpath.check(link=0, dir=1)
|
||||
|
||||
def test_symlink_remove(self):
|
||||
tmpdir = self.tmpdir.realpath()
|
||||
linkpath = tmpdir.join('test')
|
||||
linkpath.mksymlinkto(linkpath) # point to itself
|
||||
assert linkpath.check(link=1)
|
||||
linkpath.remove()
|
||||
assert not linkpath.check()
|
||||
|
||||
def test_realpath_file(self):
|
||||
tmpdir = self.tmpdir
|
||||
linkpath = tmpdir.join('test')
|
||||
filepath = tmpdir.join('file')
|
||||
filepath.write("")
|
||||
linkpath.mksymlinkto(filepath)
|
||||
realpath = linkpath.realpath()
|
||||
assert realpath.basename == 'file'
|
||||
|
||||
def test_owner(self):
|
||||
from pwd import getpwuid
|
||||
from grp import getgrgid
|
||||
stat = self.root.stat()
|
||||
assert stat.path == self.root
|
||||
|
||||
uid = stat.uid
|
||||
gid = stat.gid
|
||||
owner = getpwuid(uid)[0]
|
||||
group = getgrgid(gid)[0]
|
||||
|
||||
assert uid == stat.uid
|
||||
assert owner == stat.owner
|
||||
assert gid == stat.gid
|
||||
assert group == stat.group
|
||||
|
||||
def test_atime(self):
|
||||
import time
|
||||
path = self.root.ensure('samplefile')
|
||||
now = time.time()
|
||||
atime1 = path.atime()
|
||||
# we could wait here but timer resolution is very
|
||||
# system dependent
|
||||
path.read()
|
||||
atime2 = path.atime()
|
||||
duration = time.time() - now
|
||||
assert (atime2-atime1) <= duration
|
||||
|
||||
def test_commondir(self):
|
||||
# XXX This is here in local until we find a way to implement this
|
||||
# using the subversion command line api.
|
||||
p1 = self.root.join('something')
|
||||
p2 = self.root.join('otherthing')
|
||||
assert p1.common(p2) == self.root
|
||||
assert p2.common(p1) == self.root
|
||||
|
||||
def test_commondir_nocommon(self):
|
||||
# XXX This is here in local until we find a way to implement this
|
||||
# using the subversion command line api.
|
||||
p1 = self.root.join('something')
|
||||
p2 = py.path.local(self.root.sep+'blabla')
|
||||
assert p1.common(p2) == '/'
|
||||
|
||||
def test_join_to_root(self):
|
||||
root = self.root.parts()[0]
|
||||
assert len(str(root)) == 1
|
||||
assert str(root.join('a')) == '/a'
|
||||
|
||||
def test_join_root_to_root_with_no_abs(self):
|
||||
nroot = self.root.join('/')
|
||||
assert str(self.root) == str(nroot)
|
||||
assert self.root == nroot
|
||||
|
||||
def test_chmod_simple_int(self):
|
||||
print "self.root is", self.root
|
||||
mode = self.root.stat().mode
|
||||
self.root.chmod(mode/2)
|
||||
try:
|
||||
assert self.root.stat().mode != mode
|
||||
finally:
|
||||
self.root.chmod(mode)
|
||||
assert self.root.stat().mode == mode
|
||||
|
||||
def test_chmod_rec_int(self):
|
||||
# XXX fragile test
|
||||
print "self.root is", self.root
|
||||
recfilter = lambda x: x.check(dotfile=0, link=0)
|
||||
oldmodes = {}
|
||||
for x in self.root.visit(rec=recfilter):
|
||||
oldmodes[x] = x.stat().mode
|
||||
self.root.chmod(0772, rec=recfilter)
|
||||
try:
|
||||
for x in self.root.visit(rec=recfilter):
|
||||
assert x.stat().mode & 0777 == 0772
|
||||
finally:
|
||||
for x,y in oldmodes.items():
|
||||
x.chmod(y)
|
||||
|
||||
def test_chown_identity(self):
|
||||
owner = self.root.stat().owner
|
||||
group = self.root.stat().group
|
||||
self.root.chown(owner, group)
|
||||
|
||||
def test_chown_dangling_link(self):
|
||||
owner = self.root.stat().owner
|
||||
group = self.root.stat().group
|
||||
x = self.root.join('hello')
|
||||
x.mksymlinkto('qlwkejqwlek')
|
||||
try:
|
||||
self.root.chown(owner, group, rec=1)
|
||||
finally:
|
||||
x.remove(rec=0)
|
||||
|
||||
def test_chown_identity_rec_mayfail(self):
|
||||
owner = self.root.stat().owner
|
||||
group = self.root.stat().group
|
||||
self.root.chown(owner, group)
|
Loading…
Reference in New Issue