Commit 33c4f547 authored by Sebastian Ramacher's avatar Sebastian Ramacher

Imported Upstream version 1.0.2

parent 8cad8fdf
doc/changelog.txt
\ No newline at end of file
py/LICENSE
\ No newline at end of file
include CHANGELOG
include README.txt
include setup.py
include LICENSE
include py/LICENSE
include py/path/svn/testing/repotest.dump
include py/rest/rest.sty.template
exclude *.orig
exclude *.rej
prune .svn
prune .hg
graft doc
graft contrib
graft example
graft py/bin
graft py/rest/testing/data
graft py/misc/testing/data
Metadata-Version: 1.0
Name: py
Version: 1.0.1
Version: 1.0.2
Summary: py.test and pylib: advanced testing tool and networking lib
Home-page: http://pylib.org
Author: holger krekel, Guido Wesdorp, Carl Friedrich Bolz, Armin Rigo, Maciej Fijalkowski & others
......
......@@ -6,14 +6,4 @@ the following tools and modules:
* py.code: dynamic code generation and introspection
* py.path: uniform local and svn path objects
It includes code and contributions from several people,
listed in the LICENSE file.
For questions, please see py/doc/index.txt, refer to the website
http://pylib.org or come to the #pylib IRC freenode channel or subscribe to
http://codespeak.net/mailman/listinfo/py-dev .
have fun,
holger krekel, holger at merlinux eu
For questions and more information please visit http://pylib.org
This diff is collapsed.
import py
def pytest_runtest_call(item, __multicall__):
cap = py.io.StdCapture()
try:
return __multicall__.execute()
finally:
outerr = cap.reset()
This diff is collapsed.
"""
Allows to test twisted applications with pytest.
Notes: twisted's asynchronous behavior may have influence on the order of test-functions
TODO:
+ credits to Ralf Schmitt See: http://twistedmatrix.com/pipermail/twisted-python/2007-February/014872.html
+ get test to work
"""
import sys
try:
from twisted.internet import reactor, defer
from twisted.python import failure, log
except ImportError:
print "To use the twisted option you have to install twisted."
sys.exit(10)
try:
from greenlet import greenlet
except ImportError:
print "Since pylib 1.0 greenlet are removed and separately packaged: " \
"http://pypi.python.org/pypi/greenlet"
sys.exit(10)
def _start_twisted_logging():
"""Enables twisted internal logging"""
class Logger(object):
"""late-bound sys.stdout"""
def write(self, msg):
sys.stdout.write(msg)
def flush(self):
sys.stdout.flush()
# sys.stdout will be changed by py.test later.
log.startLogging(Logger(), setStdout=0)
def _run_twisted(logging=False):
"""Start twisted mainloop and initialize recursive calling of doit()."""
# make twisted copy traceback...
failure.Failure.cleanFailure = lambda *args: None
if logging:
_start_twisted_logging()
def fix_signal_handling():
# see http://twistedmatrix.com/trac/ticket/733
import signal
if hasattr(signal, "siginterrupt"):
signal.siginterrupt(signal.SIGCHLD, False)
def start():
fix_signal_handling()
doit(None)
# recursively called for each test-function/method due done()
def doit(val): # val always None
# switch context to wait that wrapper() passes back to test-method
res = gr_tests.switch(val)
if res is None:
reactor.stop()
return
def done(res):
reactor.callLater(0.0, doit, None) # recursive call of doit()
def err(res):
reactor.callLater(0.0, doit, res)
# the test-function *may* return a deferred
# here the test-function will actually been called
# done() is finalizing a test-process by assuring recursive invoking
# of doit()
defer.maybeDeferred(res).addCallback(done).addErrback(err)
# initially preparing the calling of doit() and starting the reactor
reactor.callLater(0.0, start)
reactor.run()
def pytest_addoption(parser):
group = parser.addgroup('twisted options')
group.addoption('--twisted-logging', action='store_true', default=False,
dest='twisted_logging',
help="switch on twisted internal logging")
def pytest_configure(config):
twisted_logging = config.getvalue("twisted_logging")
gr_twisted.switch(twisted_logging)
def pytest_unconfigure(config):
gr_twisted.switch(None)
def pytest_pyfunc_call(pyfuncitem):
# XXX1 kwargs?
# XXX2 we want to delegate actual call to next plugin
# (which may want to produce test coverage, etc.)
res = gr_twisted.switch(lambda: pyfuncitem.call())
if res:
res.raiseException()
return True # indicates that we performed the function call
gr_twisted = greenlet(_run_twisted)
gr_tests = greenlet.getcurrent()
# ===============================================================================
# plugin tests
# ===============================================================================
def test_generic(testdir):
testdir.makepyfile('''
def test_pass():
pass
from twisted.internet import defer, reactor
from twisted.python import failure
from twisted.python import log
def test_no_deferred():
assert True is True
def test_deferred():
log.msg("test_deferred() called")
d = defer.Deferred()
def done():
log.msg("test_deferred.done() CALLBACK DONE")
d.callback(None)
reactor.callLater(2.5, done)
log.msg("test_deferred() returning deferred: %r" % (d,))
return d
def test_deferred2():
log.msg("test_deferred2() called")
d = defer.Deferred()
def done():
log.msg("test_deferred2.done() CALLBACK DONE")
d.callback(None)
reactor.callLater(2.5, done)
log.msg("test_deferred2() returning deferred: %r" % (d,))
return d
def test_deferred4():
log.msg("test_deferred4() called")
from twisted.web.client import getPage
def printContents(contents):
assert contents == ""
deferred = getPage('http://twistedmatrix.com/')
deferred.addCallback(printContents)
return deferred
def test_deferred3():
log.msg("test_deferred3() called")
d = defer.Deferred()
def done():
log.msg("test_deferred3.done() CALLBACK DONE")
d.callback(None)
reactor.callLater(2.5, done)
log.msg("test_deferred3() returning deferred: %r" % (d,))
return d
class TestTwistedSetupMethod:
def setup_method(self, method):
log.msg("TestTwistedSetupMethod.setup_method() called")
def test_deferred(self):
log.msg("TestTwistedSetupMethod.test_deferred() called")
d = defer.Deferred()
def done():
log.msg("TestTwistedSetupMethod.test_deferred() CALLBACK DONE")
d.callback(None)
reactor.callLater(2.5, done)
log.msg("TestTwistedSetupMethod.test_deferred() returning deferred: %r" % (d,))
return d
def test_defer_fail():
def fun():
log.msg("provoking NameError")
rsdfg
return defer.maybeDeferred(fun)
''')
testdir.runpytest("-T")
# XXX: what to do?
# s = testdir.tmpdir.join("event.log").read()
# assert s.find("TestrunFinish") != -1
pygreen: experimental IO and execnet operations through greenlets
#!/usr/bin/env python
"""
small utility for hot-syncing a svn repository through ssh.
uses py.execnet.
"""
import py
import sys, os
def usage():
arg0 = sys.argv[0]
print """%s [user@]remote-host:/repo/location localrepo [identity keyfile]""" % (arg0,)
def main(args):
remote = args[0]
localrepo = py.path.local(args[1])
if not localrepo.check(dir=1):
raise SystemExit("localrepo %s does not exist" %(localrepo,))
if len(args) == 3:
keyfile = py.path.local(args[2])
else:
keyfile = None
remote_host, path = remote.split(':', 1)
print "ssh-connecting to", remote_host
gw = getgateway(remote_host, keyfile)
local_rev = get_svn_youngest(localrepo)
# local protocol
# 1. client sends rev/repo -> server
# 2. server checks for newer revisions and sends dumps
# 3. client receives dumps, updates local repo
# 4. client goes back to step 1
c = gw.remote_exec("""
import py
import os
remote_rev, repopath = channel.receive()
while 1:
rev = py.process.cmdexec('svnlook youngest "%s"' % repopath)
rev = int(rev)
if rev > remote_rev:
revrange = (remote_rev+1, rev)
dumpchannel = channel.gateway.newchannel()
channel.send(revrange)
channel.send(dumpchannel)
f = os.popen(
"svnadmin dump -q --incremental -r %s:%s %s"
% (revrange[0], revrange[1], repopath), 'r')
try:
maxcount = dumpchannel.receive()
count = maxcount
while 1:
s = f.read(8192)
if not s:
raise EOFError
dumpchannel.send(s)
count = count - 1
if count <= 0:
ack = dumpchannel.receive()
count = maxcount
except EOFError:
dumpchannel.close()
remote_rev = rev
else:
# using svn-hook instead would be nice here
py.std.time.sleep(30)
""")
c.send((local_rev, path))
print "checking revisions from %d in %s" %(local_rev, remote)
while 1:
revstart, revend = c.receive()
dumpchannel = c.receive()
print "receiving revisions", revstart, "-", revend, "replaying..."
svn_load(localrepo, dumpchannel)
print "current revision", revend
def svn_load(repo, dumpchannel, maxcount=100):
# every maxcount we will send an ACK to the other
# side in order to synchronise and avoid our side
# growing buffers (py.execnet does not control
# RAM usage or receive queue sizes)
dumpchannel.send(maxcount)
f = os.popen("svnadmin load -q %s" %(repo, ), "w")
count = maxcount
for x in dumpchannel:
sys.stdout.write(".")
sys.stdout.flush()
f.write(x)
count = count - 1
if count <= 0:
dumpchannel.send(maxcount)
count = maxcount
print >>sys.stdout
f.close()
def get_svn_youngest(repo):
rev = py.process.cmdexec('svnlook youngest "%s"' % repo)
return int(rev)
def getgateway(host, keyfile=None):
return py.execnet.SshGateway(host, identity=keyfile)
if __name__ == '__main__':
if len(sys.argv) < 3:
usage()
raise SystemExit(1)
main(sys.argv[1:])
"""
sysinfo.py [host1] [host2] [options]
obtain system info from remote machine.
"""
import py
import sys
optparse = py.compat.optparse
parser = optparse.OptionParser(usage=__doc__)
parser.add_option("-f", "--sshconfig", action="store", dest="ssh_config", default=None,
help="use given ssh config file, and add info all contained hosts for getting info")
parser.add_option("-i", "--ignore", action="store", dest="ignores", default=None,
help="ignore hosts (useful if the list of hostnames come from a file list)")
def parsehosts(path):
path = py.path.local(path)
l = []
rex = py.std.re.compile(r'Host\s*(\S+)')
for line in path.readlines():
m = rex.match(line)
if m is not None:
sshname, = m.groups()
l.append(sshname)
return l
class RemoteInfo:
def __init__(self, gateway):
self.gw = gateway
self._cache = {}
def exreceive(self, execstring):
if execstring not in self._cache:
channel = self.gw.remote_exec(execstring)
self._cache[execstring] = channel.receive()
return self._cache[execstring]
def getmodattr(self, modpath):
module = modpath.split(".")[0]
return self.exreceive("""
import %s
channel.send(%s)
""" %(module, modpath))
def islinux(self):
return self.getmodattr('sys.platform').find("linux") != -1
def getfqdn(self):
return self.exreceive("""
import socket
channel.send(socket.getfqdn())
""")
def getmemswap(self):
if self.islinux():
return self.exreceive("""
import commands, re
out = commands.getoutput("free")
mem = re.search(r"Mem:\s+(\S*)", out).group(1)
swap = re.search(r"Swap:\s+(\S*)", out).group(1)
channel.send((mem, swap))
""")
def getcpuinfo(self):
if self.islinux():
return self.exreceive("""
# a hyperthreaded cpu core only counts as 1, although it
# is present as 2 in /proc/cpuinfo. Counting it as 2 is
# misleading because it is *by far* not as efficient as
# two independent cores.
cpus = {}
cpuinfo = {}
f = open("/proc/cpuinfo")
lines = f.readlines()
f.close()
for line in lines + ['']:
if line.strip():
key, value = line.split(":", 1)
cpuinfo[key.strip()] = value.strip()
else:
corekey = (cpuinfo.get("physical id"),
cpuinfo.get("core id"))
cpus[corekey] = 1
numcpus = len(cpus)
model = cpuinfo.get("model name")
channel.send((numcpus, model))
""")
def debug(*args):
print >>sys.stderr, " ".join(map(str, args))
def error(*args):
debug("ERROR", args[0] + ":", *args[1:])
def getinfo(sshname, ssh_config=None, loginfo=sys.stdout):
debug("connecting to", sshname)
try:
gw = py.execnet.SshGateway(sshname, ssh_config=ssh_config)
except IOError:
error("could not get sshagteway", sshname)
else:
ri = RemoteInfo(gw)
#print "%s info:" % sshname
prefix = sshname.upper() + " "
print >>loginfo, prefix, "fqdn:", ri.getfqdn()
for attr in (
"sys.platform",
"sys.version_info",
):
loginfo.write("%s %s: " %(prefix, attr,))
loginfo.flush()
value = ri.getmodattr(attr)
loginfo.write(str(value))
loginfo.write("\n")
loginfo.flush()
memswap = ri.getmemswap()
if memswap:
mem,swap = memswap
print >>loginfo, prefix, "Memory:", mem, "Swap:", swap
cpuinfo = ri.getcpuinfo()
if cpuinfo:
numcpu, model = cpuinfo
print >>loginfo, prefix, "number of cpus:", numcpu
print >>loginfo, prefix, "cpu model", model
return ri
if __name__ == '__main__':
options, args = parser.parse_args()
hosts = list(args)
ssh_config = options.ssh_config
if ssh_config:
hosts.extend(parsehosts(ssh_config))
ignores = options.ignores or ()
if ignores:
ignores = ignores.split(",")
for host in hosts:
if host not in ignores:
getinfo(host, ssh_config=ssh_config)
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html>
<head>
<title>release-0.9.0</title>
<meta content="text/html;charset=ISO-8859-1" name="Content-Type"/>
<link href="../style.css" media="screen" rel="stylesheet" type="text/css"/></head>
<body>
<div id="navspace">
<div><a href="http://pylib.org"><img alt="py lib" height="57" id="pyimg" src="http://codespeak.net/img/pylib.png" width="77"/></a></div>
<div id="menubar">
<div style="font-style: italic;">1.0.2</div>
<div>
<div><a class="menu" href="../download.html">install</a></div></div>
<div>
<div><a class="menu" href="../contact.html">contact</a></div></div>
<div>
<div><a class="menu" href="../changelog.html">changelog</a></div></div>
<div>
<div><a class="menu" href="../faq.html">faq</a></div></div>
<div>
<div>
<h3>py.test:</h3>
<div><a class="menu" href="../test/index.html">doc index</a></div>
<div><a class="menu" href="../test/features.html">features</a></div>
<div><a class="menu" href="../test/quickstart.html">quickstart</a></div>
<div><a class="menu" href="../test/talks.html">tutorials</a></div>
<div><a class="menu" href="../test/plugin/index.html">plugins</a></div>
<div><a class="menu" href="../test/funcargs.html">funcargs</a></div>
<div><a class="menu" href="../test/customize.html">customize</a></div></div></div>
<div>
<div>
<h3>supporting APIs:</h3>
<div><a class="menu" href="../index.html">pylib index</a></div>
<div><a class="menu" href="../execnet.html">py.execnet</a></div>
<div><a class="menu" href="../path.html">py.path</a></div>
<div><a class="menu" href="../code.html">py.code</a></div></div></div></div></div>
<div id="contentspace">
<div class="document" id="py-lib-1-0-0-xxx">
<h1 class="title">py lib 1.0.0: XXX</h1>
<p>Welcome to the 1.0.0 py lib release - a library aiming to
support agile and test-driven python development on various levels.</p>
<p>XXX</p>
</div>
</div>
<script type="text/javascript">
var gaJsHost = (("https:" == document.location.protocol) ? "https://ssl." : "http://www.");
document.write(unescape("%3Cscript src='" + gaJsHost + "google-analytics.com/ga.js' type='text/javascript'%3E%3C/script%3E"));
</script>
<script type="text/javascript">
try {
var pageTracker = _gat._getTracker("UA-7597274-3");
pageTracker._trackPageview();
} catch(err) {}</script>
</body></html>
\ No newline at end of file
py lib 1.0.0: XXX
======================================================================
Welcome to the 1.0.0 py lib release - a library aiming to
support agile and test-driven python development on various levels.
XXX
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html>
<head>
<title>release-0.9.2</title>
<meta content="text/html;charset=ISO-8859-1" name="Content-Type"/>
<link href="../style.css" media="screen" rel="stylesheet" type="text/css"/></head>
<body>
<div id="navspace">
<div><a href="http://pylib.org"><img alt="py lib" height="57" id="pyimg" src="http://codespeak.net/img/pylib.png" width="77"/></a></div>
<div id="menubar">
<div style="font-style: italic;">1.0.2</div>
<div>
<div><a class="menu" href="../download.html">install</a></div></div>
<div>
<div><a class="menu" href="../contact.html">contact</a></div></div>
<div>
<div><a class="menu" href="../changelog.html">changelog</a></div></div>
<div>
<div><a class="menu" href="../faq.html">faq</a></div></div>
<div>
<div>
<h3>py.test:</h3>
<div><a class="menu" href="../test/index.html">doc index</a></div>
<div><a class="menu" href="../test/features.html">features</a></div>
<div><a class="menu" href="../test/quickstart.html">quickstart</a></div>
<div><a class="menu" href="../test/talks.html">tutorials</a></div>
<div><a class="menu" href="../test/plugin/index.html">plugins</a></div>
<div><a class="menu" href="../test/funcargs.html">funcargs</a></div>
<div><a class="menu" href="../test/customize.html">customize</a></div></div></div>
<div>
<div>
<h3>supporting APIs:</h3>
<div><a class="menu" href="../index.html">pylib index</a></div>
<div><a class="menu" href="../execnet.html">py.execnet</a></div>
<div><a class="menu" href="../path.html">py.path</a></div>
<div><a class="menu" href="../code.html">py.code</a></div></div></div></div></div>
<div id="contentspace">
<div class="document" id="py-lib-0-9-2-bugfix-release">
<h1 class="title">py lib 0.9.2: bugfix release</h1>
<p>Welcome to the 0.9.2 py lib and py.test release -
mainly fixing Windows issues, providing better
packaging and integration with setuptools.</p>
<p>Here is a quick summary of what the py lib provides:</p>
<ul class="simple">
<li>py.test: cross-project testing tool with many advanced features</li>
<li>py.execnet: ad-hoc code distribution to SSH, Socket and local sub processes</li>
<li>py.magic.greenlet: micro-threads on standard CPython (&quot;stackless-light&quot;)</li>
<li>py.path: path abstractions over local and subversion files</li>
<li>rich documentation of py's exported API</li>
<li>tested against Linux, Win32, OSX, works on python 2.3-2.6</li>
</ul>
<p>See here for more information:</p>
<p>Pypi pages: <a class="reference external" href="http://pypi.python.org/pypi/py/">http://pypi.python.org/pypi/py/</a></p>
<p>Download/Install: <a class="reference external" href="http://codespeak.net/py/0.9.2/download.html">http://codespeak.net/py/0.9.2/download.html</a></p>
<p>Documentation/API: <a class="reference external" href="http://codespeak.net/py/0.9.2/index.html">http://codespeak.net/py/0.9.2/index.html</a></p>
<p>best and have fun,</p>
<p>holger krekel</p>
</div>
</div>
<script type="text/javascript">
var gaJsHost = (("https:" == document.location.protocol) ? "https://ssl." : "http://www.");
document.write(unescape("%3Cscript src='" + gaJsHost + "google-analytics.com/ga.js' type='text/javascript'%3E%3C/script%3E"));
</script>
<script type="text/javascript">
try {
var pageTracker = _gat._getTracker("UA-7597274-3");
pageTracker._trackPageview();
} catch(err) {}</script>
</body></html>
\ No newline at end of file
py lib 0.9.2: bugfix release
=============================