Commit e37f487b authored by Sebastian Ramacher's avatar Sebastian Ramacher

Imported Upstream version 0.7-svn20050721

parents
"""\
the py lib is a development support library featuring
py.test, an interactive testing tool which supports
unit-testing with practically no boilerplate.
"""
from initpkg import initpkg
initpkg(__name__,
description = "py.test and the py lib",
revision = int('$LastChangedRevision: 14485 $'.split(':')[1][:-1]),
lastchangedate = '$LastChangedDate: 2005-07-11 05:12:29 -0300 (Mon, 11 Jul 2005) $',
version = "0.8.0-alpha2",
url = "http://codespeak.net/py",
download_url = "http://codespeak.net/download/py/py-0.8.0-alpha2.tar.gz",
license = "MIT license",
platforms = ['unix', 'linux', 'cygwin'],
author = "holger krekel & others",
author_email = "hpk@merlinux.de",
long_description = globals()['__doc__'],
exportdefs = {
'_dist.setup' : ('./misc/_dist.py', 'setup'),
# helpers for use from test functions or collectors
'test.raises' : ('./test/raises.py', 'raises'),
'test.skip' : ('./test/item.py', 'skip'),
'test.fail' : ('./test/item.py', 'fail'),
'test.exit' : ('./test/session.py', 'exit'),
'test.skip_on_error' : ('./test/item.py', 'skip_on_error'),
'test.compat.TestCase' : ('./test/compat.py', 'TestCase'),
# configuration/initialization related test api
'test.Config' : ('./test/config.py', 'Config'),
'test.ensuretemp' : ('./test/config.py', 'ensuretemp'),
'test.cmdline.main' : ('./test/cmdline.py', 'main'),
# for customization of collecting/running tests
'test.Session' : ('./test/session.py', 'Session'),
'test.TerminalSession' : ('./test/terminal/terminal.py', 'TerminalSession'),
'test.TkinterSession' : ('./test/tkinter/tksession.py', 'TkSession'),
'test.collect.Collector' : ('./test/collect.py', 'Collector'),
'test.collect.Directory' : ('./test/collect.py', 'Directory'),
'test.collect.Module' : ('./test/collect.py', 'Module'),
'test.collect.Class' : ('./test/collect.py', 'Class'),
'test.collect.Instance' : ('./test/collect.py', 'Instance'),
'test.collect.Generator' : ('./test/collect.py', 'Generator'),
'test.Item' : ('./test/item.py', 'Item'),
'test.Function' : ('./test/item.py', 'Function'),
# thread related API (still in early design phase)
'_thread.WorkerPool' : ('./thread/pool.py', 'WorkerPool'),
'_thread.NamedThreadPool' : ('./thread/pool.py', 'NamedThreadPool'),
'_thread.ThreadOut' : ('./thread/io.py', 'ThreadOut'),
# hook into the top-level standard library
'std' : ('./misc/std.py', 'std'),
'process.cmdexec' : ('./process/cmdexec.py', 'cmdexec'),
# path implementations
'path.svnwc' : ('./path/svn/wccommand.py', 'SvnWCCommandPath'),
'path.svnurl' : ('./path/svn/urlcommand.py', 'SvnCommandPath'),
'path.local' : ('./path/local/local.py', 'LocalPath'),
'path.extpy' : ('./path/extpy/extpy.py', 'Extpy'),
'path.checker' : ('./path/common.py', 'checker'),
# some nice more or less magic APIs
'magic.greenlet' : ('./magic/greenlet.py', 'greenlet'),
'magic.invoke' : ('./magic/invoke.py', 'invoke'),
'magic.revoke' : ('./magic/invoke.py', 'revoke'),
'magic.patch' : ('./magic/patch.py', 'patch'),
'magic.revert' : ('./magic/patch.py', 'revert'),
'magic.autopath' : ('./magic/autopath.py', 'autopath'),
'magic.View' : ('./magic/viewtype.py', 'View'),
'magic.AssertionError' : ('./magic/assertion.py', 'AssertionError'),
# generalized inspection API
'code.compile' : ('./code/source.py', 'compile_'),
'code.Source' : ('./code/source.py', 'Source'),
'code.Code' : ('./code/code.py', 'Code'),
'code.Frame' : ('./code/frame.py', 'Frame'),
'code.ExceptionInfo' : ('./code/excinfo.py', 'ExceptionInfo'),
'code.Traceback' : ('./code/traceback2.py', 'Traceback'),
# backports and additions of builtins
'builtin.enumerate' : ('./builtin/enumerate.py', 'enumerate'),
# gateways into remote contexts
'execnet.SocketGateway' : ('./execnet/register.py', 'SocketGateway'),
'execnet.PopenGateway' : ('./execnet/register.py', 'PopenGateway'),
'execnet.SshGateway' : ('./execnet/register.py', 'SshGateway'),
# error module, defining all errno's as Classes
'error' : ('./misc/error.py', 'error'),
# small and mean xml/html generation
'xml.html' : ('./xmlobj/html.py', 'html'),
'xml.Tag' : ('./xmlobj/xml.py', 'Tag'),
'xml.Namespace' : ('./xmlobj/xml.py', 'Namespace'),
'xml.escape' : ('./xmlobj/misc.py', 'escape'),
# logging API ('producers' and 'consumers' connected via keywords)
'log.Producer' : ('./log/producer.py', 'Producer'),
'log.default' : ('./log/producer.py', 'default'),
'log._getstate' : ('./log/producer.py', '_getstate'),
'log._setstate' : ('./log/producer.py', '_setstate'),
'log.setconsumer' : ('./log/consumer.py', 'setconsumer'),
'log.Path' : ('./log/consumer.py', 'Path'),
'log.STDOUT' : ('./log/consumer.py', 'STDOUT'),
'log.STDERR' : ('./log/consumer.py', 'STDERR'),
'compat.doctest' : ('./compat/doctest.py', '*'),
'compat.optparse' : ('./compat/optparse.py', '*'),
'compat.textwrap' : ('./compat/textwrap.py', '*'),
})
#!/usr/bin/python
#
# find and import a version of 'py'
#
import sys
import os
from os.path import dirname as opd, exists, join, basename, abspath
def searchpy(current):
while 1:
last = current
initpy = join(current, '__init__.py')
if not exists(initpy):
pydir = join(current, 'py')
# recognize py-package and ensure it is importable
if exists(pydir) and exists(join(pydir, '__init__.py')):
#for p in sys.path:
# if p == current:
# return True
if current != sys.path[0]: # if we are already first, then ok
print "inserting into sys.path:", current
sys.path.insert(0, current)
return True
current = opd(current)
if last == current:
return False
if not searchpy(abspath(os.curdir)):
if not searchpy(opd(abspath(sys.argv[0]))):
if not searchpy(opd(__file__)):
pass # let's hope it is just on sys.path
import py
if __name__ == '__main__':
print "py lib is at", py.__file__
#!/usr/bin/python
from _findpy import py
import sys
pydir = py.path.local(py.__file__).dirpath()
rootdir = pydir.dirpath()
def gen_manifest():
pywc = py.path.svnwc(pydir)
status = pywc.status(rec=True)
#assert not status.modified
#assert not status.deleted
#assert not status.added
versioned = dict([(x.localpath,1) for x in status.allpath()])
l = []
for x in rootdir.visit(None, lambda x: x.basename != '.svn'):
if x.check(file=1):
names = [y.basename for y in x.parts()]
if '.svn' in names:
l.append(x)
elif x in versioned:
l.append(x)
l.append(rootdir / "setup.py")
l = [x.relto(rootdir) for x in l]
l.append("")
s = "\n".join(l)
return s
def trace(arg):
lines = str(arg).split('\n')
prefix = "[trace] "
prefix = "* "
indent = len(prefix)
ispace = " " * indent
lines = [ispace + line for line in lines]
if lines:
lines[0] = prefix + lines[0][indent:]
for line in lines:
print >>py.std.sys.stdout, line
def make_distfiles(tmpdir):
""" return distdir with tar.gz and zipfile. """
manifest = tmpdir.join('MANIFEST')
trace("generating %s" %(manifest,))
content = gen_manifest()
manifest.write(content)
trace("wrote %d files into manifest file" %len(content.split('\n')))
distdir = tmpdir.ensure('dist', dir=1)
oldir = rootdir.chdir()
try:
trace("invoking sdist, generating into %s" % (distdir,))
py._dist.setup(py, script_name="setup.py",
script_args=('-q', 'sdist', '--no-prune',
'-m', str(manifest),
'--formats=gztar,zip',
'-d', str(distdir)))
py._dist.setup(py, script_name="setup.py",
script_args=('-q', 'bdist_wininst',
#'-m', str(manifest),
'-d', str(distdir)))
finally:
oldir.chdir()
return distdir
def pytest(unpacked):
trace("py-testing %s" % unpacked)
old = unpacked.chdir()
try:
import os
os.system("python py/bin/py.test py")
finally:
old.chdir()
def unpackremotetar(tmpdir, strurl):
import tarfile, urllib
f = urllib.urlopen(strurl)
basename = strurl.split('/')[-1]
target = tmpdir.join(basename)
trace("downloading to %s" %(target,))
target.write(f.read())
trace("extracting to %s" %(target,))
old = tmpdir.chdir()
try:
py.process.cmdexec("tar zxf %s" %(target,))
finally:
old.chdir()
prefix = '.tar.gz'
assert basename.endswith(prefix)
stripped = basename[:-len(prefix)]
unpacked = tmpdir.join(stripped)
assert unpacked
return unpacked
def checksvnworks(unpacked):
pywc = py.path.svnwc(unpacked.join('py'))
trace("checking versioning works: %s" %(pywc,))
status = pywc.status(rec=True)
assert not status.modified
assert not status.deleted
assert not status.unknown
def pytest_remote(address, url):
gw = py.execnet.SshGateway(address)
basename = url[url.rfind('/')+1:]
purebasename = basename[:-len('.tar.gz')]
def mytrace(x, l=[]):
l.append(x)
if x.endswith('\n'):
trace("".join(l))
l[:] = []
channel = gw.remote_exec(stdout=mytrace, stderr=sys.stderr, source="""
url = %(url)r
basename = %(basename)r
purebasename = %(purebasename)r
import os, urllib
f = urllib.urlopen(url)
print "reading from", url
s = f.read()
f.close()
f = open(basename, 'w')
f.write(s)
f.close()
if os.path.exists(purebasename):
import shutil
shutil.rmtree(purebasename)
os.system("tar zxf %%s" %% (basename,))
print "unpacked", purebasename
os.chdir(purebasename)
print "testing at %(address)s ..."
#os.system("python py/bin/py.test py")
import commands
status, output = commands.getstatusoutput("python py/bin/py.test py")
#print output
print "status:", status
""" % locals())
channel.waitclose(200.0)
if __name__ == '__main__':
py.magic.invoke(assertion=True)
version = py.std.sys.argv[1]
assert py.__package__.version == version, (
"py package has version %s\nlocation: %s" %
(py.__package__.version, pydir))
tmpdir = py.path.local.get_temproot().join('makepyrelease-%s' % version)
if tmpdir.check():
trace("removing %s" %(tmpdir,))
tmpdir.remove()
tmpdir.mkdir()
trace("using tmpdir %s" %(tmpdir,))
distdir = make_distfiles(tmpdir)
targz = distdir.join('py-%s.tar.gz' % version)
zip = distdir.join('py-%s.zip' % version)
files = distdir.listdir()
for fn in files:
assert fn.check(file=1)
remotedir = 'codespeak.net://www/codespeak.net/htdocs/download/py/'
source = distdir # " ".join([str(x) for x in files])
trace("rsyncing %(source)s to %(remotedir)s" % locals())
py.process.cmdexec("rsync -avz %(source)s/ %(remotedir)s" % locals())
ddir = tmpdir.ensure('download', dir=1)
URL = py.__package__.download_url # 'http://codespeak.net/download/py/'
unpacked = unpackremotetar(ddir, URL)
assert unpacked == ddir.join("py-%s" % (version,))
#checksvnworks(unpacked)
#pytest(unpacked)
pytest_remote('test@codespeak.net', py.__package__.download_url)
#!/usr/bin/env python
from _findpy import py
import py
for x in py.path.local().visit('*.pyc', py.path.checker(dotfile=0)):
x.remove()
#!/usr/bin/env python
# hands on script to compute the non-empty Lines of Code
# for tests and non-test code
import py
curdir = py.path.local()
class Filecounter:
def __init__(self):
self.counts = {}
def count(self, fn, empty=False):
if empty:
s = len(p.readlines())
else:
s = 0
for i in fn.readlines():
if i.strip():
s += 1
self.counts[fn] = s
if __name__ == '__main__':
testcounter = Filecounter()
counter = Filecounter()
args = py.std.sys.argv[1:]
if not args:
args = ['.']
locations = [py.path.local(x) for x in args]
for loc in locations:
for x in loc.visit('*.py', py.path.checker(dotfile=0)):
bn = x.basename
if bn.startswith('test_') or bn.endswith('_test.py'):
testcounter.count(x)
else:
counter.count(x)
numfiles = len(counter.counts)
numtestfiles = len(testcounter.counts)
numlines = sum(counter.counts.values())
numtestlines = sum(testcounter.counts.values())
#for x,y in counter.counts.items():
# print "%3d %30s" % (y,x)
items = counter.counts.items()
items.sort(lambda x,y: cmp(x[1], y[1]))
for x, y in items:
print "%3d %30s" % (y,x)
print "%30s %3d" %("number of testfiles", numtestfiles)
print "%30s %3d" %("number of testlines", numtestlines)
print "%30s %3d" %("number of files", numfiles)
print "%30s %3d" %("number of lines", numlines)
#!/usr/bin/python
from _findpy import py
import re
string = py.std.sys.argv[1]
curdir = py.path.local()
for x in curdir.visit('*.py', py.path.checker(dotfile=0)):
s = x.read()
if s.find(string) != -1:
lines = x.readlines()
for i, line in enumerate(lines):
if line.find(string) != -1:
print "%s:%d %s" %(x.relto(curdir), i+1, line.rstrip())
#!/usr/bin/python
"""
invoke
rest.py filename1.txt directory
to generate html files from restructered text.
http://docutils.sourceforge.net/docs/user/rst/quickref.html
"""
import os, sys
from _findpy import py
from py.__.misc import rest
if __name__=='__main__':
basedir = os.path.dirname(os.path.dirname(os.path.abspath(sys.argv[0])))
sys.path.insert(0, basedir)
if len(sys.argv) == 1:
filenames = [py.path.svnwc()]
else:
filenames = [py.path.svnwc(x) for x in sys.argv[1:]]
for p in filenames:
if not p.check():
log("path %s not found, ignoring" % p)
continue
if p.check(dir=1):
for x in p.visit(fil=py.path.checker(fnmatch='*.txt', versioned=True),
rec=py.path.checker(dotfile=0)):
rest.process(x)
elif p.check(file=1):
rest.process(p)
#!/usr/bin/env python
from _findpy import py
py.test.cmdline.main()
@echo off
python %~dp0\py.test %*
@echo off
python %~dp0\..\py.cleanup %*
\ No newline at end of file
@echo off
python %~dp0\..\py.countloc %*
\ No newline at end of file
@echo off
python %~dp0\..\py.lookup %*
\ No newline at end of file
@echo off
python %~dp0\..\py.rest %*
\ No newline at end of file
@echo off
python %~dp0\..\py.test %*
\ No newline at end of file
from __future__ import generators
try:
enumerate = enumerate
except NameError:
def enumerate(iterable):
i = 0
for x in iterable:
yield i, x
i += 1
import sys
import py
def test_enumerate():
l = [0,1,2]
for i,x in py.builtin.enumerate(l):
assert i == x
import py
class Directory(py.test.collect.Directory):
# XXX see in which situations/platforms we want
# run tests here
#def recfilter(self, path):
# if py.std.sys.platform == 'linux2':
# if path.basename == 'greenlet':
# return False
# return super(Directory, self).recfilter(path)
def run(self):
py.test.skip("c-extension testing needs platform selection")
This code is by Armin Rigo with the core pieces by Christian Tismer.
These pieces are:
- slp_platformselect.h
- switch_*.h, included from the previous one.
All additional credits for the general idea of stack switching also go to
Christian. In other words, if it works it is thanks to Christian's work,
and if it crashes it is because of a bug of mine :-)
This code is going to be integrated in the 'py' lib. The test file only runs
if you got greenlets from there! See http://codespeak.net/py/ and try
'from py.magic import greenlet'.
-- Armin
import thread, sys
__all__ = ['greenlet', 'main', 'getcurrent']
class greenlet(object):
__slots__ = ('run', '_controller')
def __init__(self, run=None, parent=None):
if run is not None:
self.run = run
if parent is not None:
self.parent = parent
def switch(self, *args):
global _passaround_
_passaround_ = None, args, None
self._controller.switch(self)
exc, val, tb = _passaround_
del _passaround_
if exc is None:
if isinstance(val, tuple) and len(val) == 1:
return val[0]
else:
return val
else:
raise exc, val, tb
def __nonzero__(self):
return self._controller.isactive()
def __new__(cls, *args, **kwds):
self = object.__new__(cls)
self._controller = _Controller()
return self
def __del__(self):
#print 'DEL:', self
if self._controller.parent is None:
return # don't kill the main greenlet
while self._controller.isactive():
self._controller.kill(self)
def getparent(self):
return self._controller.parent
def setparent(self, nparent):
if not isinstance(nparent, greenlet):
raise TypeError, "parent must be a greenlet"
p = nparent
while p is not None:
if p is self:
raise ValueError, "cyclic parent chain"
p = p._controller.parent
self._controller.parent = nparent
parent = property(getparent, setparent)
del getparent
del setparent
class _Controller:
# Controllers are separated from greenlets to allow greenlets to be
# deallocated while running, when their last reference goes away.
# Care is taken to keep only references to controllers in thread's
# frames' local variables.
# _Controller.parent: the parent greenlet.