jiuyuan
/
test_ok1
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

* fix some syntax and 3k issues for py/path and py/process, tests only partially working 

* have py.process.cmdexec return unicode/text (for now)
* rename py.builtin.basestring to _basestring

--HG--
branch : trunk
This commit is contained in:
holger krekel 2009-08-29 16:40:03 +02:00
parent 0f7a9e2da2
commit b930565d56
13 changed files with 98 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
py/__init__.py
View File

@ -147,7 +147,7 @@ initpkg(__name__,
'builtin.print_' : ('./builtin/builtin31.py', 'print_'),
'builtin._reraise' : ('./builtin/builtin31.py', '_reraise'),
'builtin.exec_' : ('./builtin/builtin31.py', 'exec_'),
'builtin.basestring' : ('./builtin/builtin31.py', 'basestring'),
'builtin._basestring' : ('./builtin/builtin31.py', '_basestring'),
'builtin._totext' : ('./builtin/builtin31.py', '_totext'),
'builtin.builtins' : ('./builtin/builtin31.py', 'builtins'),

7
py/builtin/builtin31.py
View File

@ -3,7 +3,9 @@ import sys
if sys.version_info >= (3, 0):
exec ("print_ = print ; exec_=exec")
import builtins
basestring = str
# some backward compatibility helpers
_basestring = str
def _totext(obj, encoding):
if isinstance(obj, str):
obj = obj.encode(encoding)
@ -11,7 +13,8 @@ if sys.version_info >= (3, 0):
else:
_totext = unicode
basestring = basestring
_basestring = basestring
import __builtin__ as builtins
def print_(*args, **kwargs):
""" minimal backport of py3k print statement. """

2
py/code/source.py
View File

@ -25,7 +25,7 @@ class Source(object):
partlines = []
if isinstance(part, Source):
partlines = part.lines
elif isinstance(part, py.builtin.basestring):
elif isinstance(part, py.builtin._basestring):
partlines = part.split('\n')
if rstrip:
while partlines:

2
py/path/local.py
View File

@ -133,7 +133,7 @@ class LocalPath(FSBase):
self = object.__new__(cls)
if not path:
self.strpath = os.getcwd()
elif isinstance(path, str):
elif isinstance(path, py.builtin._basestring):
self.strpath = os.path.abspath(os.path.normpath(str(path)))
else:
raise ValueError("can only pass None, Path instances "

10
py/path/svnurl.py
View File

@ -56,7 +56,7 @@ class SvnCommandPath(svncommon.SvnPathBase):
# fixing the locale because we can't otherwise parse
string = " ".join(l)
if DEBUG:
print "execing", string
print("execing %s" % string)
out = self._svncmdexecauth(string)
return out
@ -70,7 +70,8 @@ class SvnCommandPath(svncommon.SvnPathBase):
def _cmdexec(self, cmd):
try:
out = process.cmdexec(cmd)
except py.process.cmdexec.Error, e:
except py.process.cmdexec.Error:
e = sys.exc_info()[1]
if (e.err.find('File Exists') != -1 or
e.err.find('File already exists') != -1):
raise py.error.EEXIST(self)
@ -207,7 +208,7 @@ checkin message msg."""
def _proplist(self):
res = self._svnwithrev('proplist')
lines = res.split('\n')
lines = map(str.strip, lines[1:])
lines = [x.strip() for x in lines[1:]]
return svncommon.PropListDict(self, lines)
def _listdir_nameinfo(self):
@ -215,7 +216,8 @@ checkin message msg."""
def builder():
try:
res = self._svnwithrev('ls', '-v')
except process.cmdexec.Error, e:
except process.cmdexec.Error:
e = sys.exc_info()[1]
if e.err.find('non-existent in that revision') != -1:
raise py.error.ENOENT(self, e.err)
elif e.err.find('File not found') != -1:

45
py/path/svnwc.py
View File

@ -201,7 +201,7 @@ class SvnPathBase(common.PathBase):
elif name == 'ext':
res.append(ext)
else:
raise NameError, "Don't know part %r" % name
raise NameError("Don't know part %r" % name)
return res
def __eq__(self, other):
@ -251,7 +251,7 @@ class SvnPathBase(common.PathBase):
if fil or sort:
paths = filter(fil, paths)
paths = isinstance(paths, list) and paths or list(paths)
if callable(sort):
if hasattr(sort, '__call__'):
paths.sort(sort)
elif sort:
paths.sort()
@ -345,7 +345,7 @@ class SvnPathBase(common.PathBase):
def parse_apr_time(timestr):
i = timestr.rfind('.')
if i == -1:
raise ValueError, "could not parse %s" % timestr
raise ValueError("could not parse %s" % timestr)
timestr = timestr[:i]
parsedtime = time.strptime(timestr, "%Y-%m-%dT%H:%M:%S")
return time.mktime(parsedtime)
@ -469,7 +469,7 @@ class SvnWCCommandPath(common.PathBase):
if getattr(self, '_url', None) is None:
info = self.info()
self._url = info.url #SvnPath(info.url, info.rev)
assert isinstance(self._url, str)
assert isinstance(self._url, py.builtin._basestring)
return self._url
url = property(_geturl, None, None, "url of this WC item")
@ -520,7 +520,8 @@ class SvnWCCommandPath(common.PathBase):
os.environ[key] = hold
else:
del os.environ[key]
except py.process.cmdexec.Error, e:
except py.process.cmdexec.Error:
e = sys.exc_info()[1]
strerr = e.err.lower()
if strerr.find('file not found') != -1:
raise py.error.ENOENT(self)
@ -577,7 +578,6 @@ class SvnWCCommandPath(common.PathBase):
a file). if you specify a keyword argument 'directory=True'
then the path is forced to be a directory path.
"""
try:
p = self.join(*args)
if p.check():
if p.check(versioned=False):
@ -590,8 +590,6 @@ class SvnWCCommandPath(common.PathBase):
p.write("")
p.add()
return p
except:
error_enhance(sys.exc_info())
def mkdir(self, *args):
""" create & return the directory joined with args. """
@ -762,7 +760,7 @@ If rec is True, then return a dictionary mapping sub-paths to such mappings.
else:
res = self._svn('proplist')
lines = res.split('\n')
lines = map(str.strip, lines[1:])
lines = [x.strip() for x in lines[1:]]
return PropListDict(self, lines)
def revert(self, rec=0):
@ -806,7 +804,8 @@ recursively. """
if not info:
try:
output = self._svn('info')
except py.process.cmdexec.Error, e:
except py.process.cmdexec.Error:
e = sys.exc_info()[1]
if e.err.find('Path is not a working copy directory') != -1:
raise py.error.ENOENT(self, e.err)
elif e.err.find("is not under version control") != -1:
@ -849,7 +848,7 @@ recursively. """
if fil or sort:
paths = filter(fil, paths)
paths = isinstance(paths, list) and paths or list(paths)
if callable(sort):
if hasattr(sort, '__call__'):
paths.sort(sort)
elif sort:
paths.sort()
@ -871,7 +870,8 @@ recursively. """
s = self.svnwcpath.info()
except (py.error.ENOENT, py.error.EEXIST):
return False
except py.process.cmdexec.Error, e:
except py.process.cmdexec.Error:
e = sys.exc_info()[1]
if e.err.find('is not a working copy')!=-1:
return False
if e.err.lower().find('not a versioned resource') != -1:
@ -1007,7 +1007,7 @@ class WCStatus:
# because of the way SVN presents external items
continue
# keep trying
raise ValueError, "could not parse line %r" % line
raise ValueError("could not parse line %r" % line)
else:
rev, modrev, author, fn = m.groups()
wcpath = rootwcpath.join(fn, abs=1)
@ -1065,7 +1065,8 @@ class XMLWCStatus(WCStatus):
minidom, ExpatError = importxml()
try:
doc = minidom.parseString(data)
except ExpatError, e:
except ExpatError:
e = sys.exc_info()[1]
raise ValueError(str(e))
urevels = doc.getElementsByTagName('against')
if urevels:
@ -1179,7 +1180,7 @@ class InfoSvnWCCommand:
try:
self.url = d['url']
except KeyError:
raise ValueError, "Not a versioned resource"
raise ValueError("Not a versioned resource")
#raise ValueError, "Not a versioned resource %r" % path
self.kind = d['nodekind'] == 'directory' and 'dir' or d['nodekind']
self.rev = int(d['revision'])
@ -1201,7 +1202,7 @@ def parse_wcinfotime(timestr):
# example: 2003-10-27 20:43:14 +0100 (Mon, 27 Oct 2003)
m = re.match(r'(\d+-\d+-\d+ \d+:\d+:\d+) ([+-]\d+) .*', timestr)
if not m:
raise ValueError, "timestring %r does not match" % timestr
raise ValueError("timestring %r does not match" % timestr)
timestr, timezone = m.groups()
# do not handle timezone specially, return value should be UTC
parsedtime = time.strptime(timestr, "%Y-%m-%d %H:%M:%S")
@ -1217,7 +1218,7 @@ def make_recursive_propdict(wcroot,
line = lines.pop(0)
m = rex.match(line)
if not m:
raise ValueError, "could not parse propget-line: %r" % line
raise ValueError("could not parse propget-line: %r" % line)
path = m.groups()[0]
wcpath = wcroot.join(path, abs=1)
propnames = []
@ -1228,8 +1229,6 @@ def make_recursive_propdict(wcroot,
pdict[wcpath] = PropListDict(wcpath, propnames)
return pdict
def error_enhance((cls, error, tb)):
raise cls, error, tb
def importxml(cache=[]):
if cache:
@ -1244,18 +1243,18 @@ class LogEntry:
self.rev = int(logentry.getAttribute('revision'))
for lpart in filter(None, logentry.childNodes):
if lpart.nodeType == lpart.ELEMENT_NODE:
if lpart.nodeName == u'author':
if lpart.nodeName == 'author':
self.author = lpart.firstChild.nodeValue.encode('UTF-8')
elif lpart.nodeName == u'msg':
elif lpart.nodeName == 'msg':
if lpart.firstChild:
self.msg = lpart.firstChild.nodeValue.encode('UTF-8')
else:
self.msg = ''
elif lpart.nodeName == u'date':
elif lpart.nodeName == 'date':
#2003-07-29T20:05:11.598637Z
timestr = lpart.firstChild.nodeValue.encode('UTF-8')
self.date = parse_apr_time(timestr)
elif lpart.nodeName == u'paths':
elif lpart.nodeName == 'paths':
self.strpaths = []
for ppart in filter(None, lpart.childNodes):
if ppart.nodeType == ppart.ELEMENT_NODE:

18
py/path/testing/svntestbase.py
View File

@ -3,6 +3,7 @@ import py
from py import path, test, process
from py.__.path.testing.fscommon import CommonFSTests, setuptestfs
from py.__.path import svnwc as svncommon
from py.builtin import print_
repodump = py.path.local(__file__).dirpath('repotest.dump')
@ -26,15 +27,15 @@ def getrepowc(reponame='basetestrepo', wcname='wc'):
svncommon._escape_helper(repo))
py.process.cmdexec('svnadmin load -q "%s" <"%s"' %
(svncommon._escape_helper(repo), repodump))
print "created svn repository", repo
print_("created svn repository", repo)
wcdir.ensure(dir=1)
wc = py.path.svnwc(wcdir)
if py.std.sys.platform == 'win32':
repo = '/' + str(repo).replace('\\', '/')
wc.checkout(url='file://%s' % repo)
print "checked out new repo into", wc
print_("checked out new repo into", wc)
else:
print "using repository at", repo
print_("using repository at", repo)
wc = py.path.svnwc(wcdir)
return ("file://%s" % repo, wc)
@ -49,12 +50,13 @@ def save_repowc():
wc.localpath.copy(savedwc.localpath)
return savedrepo, savedwc
def restore_repowc((savedrepo, savedwc)):
def restore_repowc(obj):
savedrepo, savedwc = obj
repo, wc = getrepowc()
print repo
print repo[len("file://"):]
print (repo)
print (repo[len("file://"):])
repo = py.path.local(repo[len("file://"):])
print repo
print (repo)
assert repo.check()
# repositories have read only files on windows
#repo.chmod(0777, rec=True)
@ -79,7 +81,7 @@ def make_test_repo(name="test-repository"):
class CommonSvnTests(CommonFSTests):
def setup_method(self, meth):
bn = meth.func_name
bn = meth.__name__
for x in 'test_remove', 'test_move', 'test_status_deleted':
if bn.startswith(x):
self._savedrepowc = save_repowc()

1
py/path/testing/test_svnurl.py
View File

@ -78,7 +78,6 @@ class TestSvnURLCommandPath(CommonCommandAndBindingTests):
bar = foo.join('bar').ensure(file=True)
bar.write('bar\n')
rev1 = foo.commit('testing something')
print 'rev1:', rev1
baz = foo.join('baz').ensure(file=True)
baz.write('baz\n')
rev2 = foo.commit('testing more')

2
py/path/testing/test_svnwc.py
View File

@ -247,7 +247,7 @@ class TestWCSvnCommandPath(CommonSvnTests):
def test_status_wrong_xml(self):
# testing for XML without author - this used to raise an exception
xml = u'<entry path="/home/jean/zope/venv/projectdb/parts/development-products/DataGridField">\n<wc-status item="incomplete" props="none" revision="784">\n</wc-status>\n</entry>'
xml = '<entry path="/home/jean/zope/venv/projectdb/parts/development-products/DataGridField">\n<wc-status item="incomplete" props="none" revision="784">\n</wc-status>\n</entry>'
st = XMLWCStatus.fromstring(xml, self.root)
assert len(st.incomplete) == 1

24
py/process/cmdexec.py
View File

@ -23,7 +23,7 @@ def posix_exec_cmd(cmd):
the exception will provide an 'err' attribute containing
the error-output from the command.
"""
__tracebackhide__ = True
#__tracebackhide__ = True
try:
from subprocess import Popen, PIPE
except ImportError:
@ -31,7 +31,6 @@ def posix_exec_cmd(cmd):
import errno
#print "execing", cmd
child = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE,
close_fds=True)
stdin, stdout, stderr = child.stdin, child.stdout, child.stderr
@ -53,12 +52,13 @@ def posix_exec_cmd(cmd):
import select
out, err = [], []
while 1:
r_list = filter(lambda x: x and not x.closed, [stdout, stderr])
r_list = [x for x in [stdout, stderr] if x and not x.closed]
if not r_list:
break
try:
r_list = select.select(r_list, [], [])[0]
except (select.error, IOError), se:
except (select.error, IOError):
se = sys.exc_info()[1]
if se.args[0] == errno.EINTR:
continue
else:
@ -66,12 +66,14 @@ def posix_exec_cmd(cmd):
for r in r_list:
try:
data = r.read() # XXX see XXX above
except IOError, io:
except IOError:
io = sys.exc_info()[1]
if io.args[0] == errno.EAGAIN:
continue
# Connection Lost
raise
except OSError, ose:
except OSError:
ose = sys.exc_info()[1]
if ose.errno == errno.EPIPE:
# Connection Lost
raise
@ -88,15 +90,19 @@ def posix_exec_cmd(cmd):
err.append(data)
pid, systemstatus = os.waitpid(child.pid, 0)
if pid != child.pid:
raise ExecutionFailed, "child process disappeared during: "+ cmd
raise ExecutionFailed("child process disappeared during: "+ cmd)
if systemstatus:
if os.WIFSIGNALED(systemstatus):
status = os.WTERMSIG(systemstatus) + 128
else:
status = os.WEXITSTATUS(systemstatus)
raise ExecutionFailed(status, systemstatus, cmd,
''.join(out), ''.join(err))
return "".join(out)
joiner(out), joiner(err))
return joiner(out)
def joiner(out):
encoding = sys.getdefaultencoding()
return "".join([py.builtin._totext(x, encoding) for x in out])
#-----------------------------------------------------------
# simple win32 external command execution

11
py/process/testing/test_cmdexec.py
View File

@ -1,6 +1,9 @@
import py
from py.process import cmdexec
def exvalue():
return py.std.sys.exc_info()[1]
class Test_exec_cmd:
def test_simple(self):
out = cmdexec('echo hallo')
@ -12,14 +15,16 @@ class Test_exec_cmd:
def test_simple_error_exact_status(self):
try:
cmdexec('exit 1')
except cmdexec.Error, e:
except cmdexec.Error:
e = exvalue()
assert e.status == 1
def test_err(self):
try:
cmdexec('echoqweqwe123 hallo')
raise AssertionError, "command succeeded but shouldn't"
except cmdexec.Error, e:
raise AssertionError("command succeeded but shouldn't")
except cmdexec.Error:
e = exvalue()
assert hasattr(e, 'err')
assert hasattr(e, 'out')
assert e.err or e.out

29
py/process/testing/test_forkedfunc.py
View File

@ -69,7 +69,7 @@ def test_forkedfunc_huge_data():
def test_box_seq():
# we run many boxes with huge data, just one after another
for i in xrange(50):
for i in range(50):
result = py.process.ForkedFunc(boxhuge).waitfinish()
assert result.out
assert result.exitstatus == 0
@ -79,8 +79,8 @@ def test_box_seq():
def test_box_in_a_box():
def boxfun():
result = py.process.ForkedFunc(boxf2).waitfinish()
print result.out
print >>sys.stderr, result.err
print (result.out)
sys.stderr.write(result.err + "\n")
return result.retval
result = py.process.ForkedFunc(boxfun).waitfinish()
@ -114,25 +114,26 @@ def test_kill_func_forked():
#
def boxf1():
print "some out"
print >>sys.stderr, "some err"
sys.stdout.write("some out\n")
sys.stderr.write("some err\n")
return 1
def boxf2():
os.write(1, "someout")
os.write(2, "someerr")
os.write(1, "someout".encode('ascii'))
os.write(2, "someerr".encode('ascii'))
return 2
def boxseg():
os.kill(os.getpid(), 11)
def boxhuge():
os.write(1, " " * 10000)
os.write(2, " " * 10000)
os.write(1, " " * 10000)
s = " ".encode('ascii')
os.write(1, s * 10000)
os.write(2, s * 10000)
os.write(1, s * 10000)
os.write(1, " " * 10000)
os.write(2, " " * 10000)
os.write(2, " " * 10000)
os.write(1, " " * 10000)
os.write(1, s * 10000)
os.write(2, s * 10000)
os.write(2, s * 10000)
os.write(1, s * 10000)
return 3

2
py/test/pycollect.py
View File

@ -305,7 +305,7 @@ class Generator(FunctionMixin, PyCollectorMixin, py.test.collect.Collector):
if not isinstance(obj, (tuple, list)):
obj = (obj,)
# explict naming
if isinstance(obj[0], basestring):
if isinstance(obj[0], py.builtin._basestring):
name = obj[0]
obj = obj[1:]
else: