Commit 95eb8bbe authored by Sebastian Ramacher's avatar Sebastian Ramacher

Imported Upstream version 1.1.1

parent 33c4f547
# Automatically generated by `hgimportsvn`
syntax:glob
.svn
.hgsvn
# These lines are suggested according to the svn:ignore property
# Feel free to enable them by uncommenting them
syntax:glob
*.pyc
*.pyo
*.swp
*.html
*.class
*.orig
build/
dist/
py.egg-info
issue/
3rdparty/
doc/changelog.txt
\ No newline at end of file
This diff is collapsed.
py/LICENSE
\ No newline at end of file
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
include CHANGELOG
include README.txt
include setup.py
include distribute_setup.py
include LICENSE
include py/LICENSE
include py/path/svn/testing/repotest.dump
include py/rest/rest.sty.template
include conftest.py
graft doc
graft contrib
graft bin
graft testing
exclude *.orig
exclude *.rej
exclude .hginore
exclude *.pyc
recursive-exclude testing *.pyc *.orig *.rej *$py.class
prune .pyc
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.2
Summary: py.test and pylib: advanced testing tool and networking lib
Version: 1.1.1
Summary: py.test and pylib: rapid testing and development utils.
Home-page: http://pylib.org
Author: holger krekel, Guido Wesdorp, Carl Friedrich Bolz, Armin Rigo, Maciej Fijalkowski & others
Author-email: holger at merlinux.eu, py-dev at codespeak.net
Author-email: holger at merlinux.eu
License: MIT license
Description:
advanced testing and development support library:
py.test and pylib: rapid testing and development utils
- `py.test`_: cross-project testing tool with many advanced features
- `py.execnet`_: ad-hoc code distribution to SSH, Socket and local sub processes
- `py.path`_: path abstractions over local and subversion files
- `py.code`_: dynamic code compile and traceback printing support
Compatibility: Linux, Win32, OSX, Python versions 2.3-2.6.
Platforms: Linux, Win32, OSX
Interpreters: Python versions 2.4 through to 3.1, Jython 2.5.1.
For questions please check out http://pylib.org/contact.html
.. _`py.test`: http://pylib.org/test.html
.. _`py.execnet`: http://pylib.org/execnet.html
.. _`py.path`: http://pylib.org/path.html
.. _`py.code`: http://pylib.org/code.html
(c) Holger Krekel and others, 2009
Platform: unix
Platform: linux
......@@ -38,6 +36,5 @@ Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Topic :: Software Development :: Testing
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Classifier: Topic :: Utilities
Classifier: Programming Language :: Python
......@@ -2,7 +2,6 @@ The py lib is a Python development support library featuring
the following tools and modules:
* py.test: tool for distributed automated testing
* py.execnet: ad-hoc distributed execution
* py.code: dynamic code generation and introspection
* py.path: uniform local and svn path objects
......
......@@ -19,7 +19,7 @@ def searchpy(current):
# if p == current:
# return True
if current != sys.path[0]: # if we are already first, then ok
print >>sys.stderr, "inserting into sys.path:", current
sys.stderr.write("inserting into sys.path: %s\n" % current)
sys.path.insert(0, current)
return True
current = opd(current)
......@@ -34,4 +34,4 @@ if not searchpy(abspath(os.curdir)):
import py
if __name__ == '__main__':
print "py lib is at", py.__file__
print ("py lib is at %s" % py.__file__)
File added
@echo off
for /F "usebackq delims=" %%i in (`python "%~dp0\env.py"`) do %%i
......@@ -3,7 +3,7 @@
import sys, os, os.path
progpath = sys.argv[0]
packagedir = os.path.abspath(os.path.dirname(progpath))
packagedir = os.path.dirname(os.path.dirname(os.path.abspath(progpath)))
packagename = os.path.basename(packagedir)
bindir = os.path.join(packagedir, 'bin')
if sys.platform == 'win32':
......@@ -29,5 +29,5 @@ def setenv(name, value):
assert False, 'Shell not supported.'
return cmd
print prepend_path('PATH', bindir)
print prepend_path('PYTHONPATH', rootdir)
print(prepend_path('PATH', bindir))
print(prepend_path('PYTHONPATH', rootdir))
#!/usr/bin/env python
from _findpy import py
py.cmdline.pyrest()
\ No newline at end of file
py.cmdline.pyconvert_unittest()
\ No newline at end of file
@echo off
python "%~dp0\..\py.convert_unittest" %*
\ No newline at end of file
import py
pytest_plugins = '_pytest doctest pytester'.split()
rsyncdirs = ['../doc']
rsyncdirs = ['conftest.py', 'bin', 'py', 'doc', 'testing']
try:
import execnet
except ImportError:
pass
else:
rsyncdirs.append(str(py.path.local(execnet.__file__).dirpath()))
import py
def pytest_addoption(parser):
group = parser.addgroup("pylib", "py lib testing options")
group = parser.getgroup("pylib", "py lib testing options")
group.addoption('--sshhost',
action="store", dest="sshhost", default=None,
help=("ssh xspec for ssh functional tests. "))
......@@ -17,15 +26,18 @@ def pytest_addoption(parser):
def pytest_funcarg__specssh(request):
return getspecssh(request.config)
def pytest_funcarg__specsocket(request):
return getsocketspec(request.config)
def getgspecs(config=None):
if config is None:
config = py.test.config
return [execnet.XSpec(spec)
for spec in config.getvalueorskip("gspecs")]
# configuration information for tests
def getgspecs(config=None):
if config is None:
config = py.test.config
return [py.execnet.XSpec(spec)
return [execnet.XSpec(spec)
for spec in config.getvalueorskip("gspecs")]
def getspecssh(config=None):
......@@ -49,7 +61,7 @@ def pytest_generate_tests(metafunc):
multi = getattr(metafunc.function, 'multi', None)
if multi is None:
return
assert len(multi.__dict__) == 1
for name, l in multi.__dict__.items():
assert len(multi.kwargs) == 1
for name, l in multi.kwargs.items():
for val in l:
metafunc.addcall(funcargs={name: val})
This diff is collapsed.
"""
this little helper allows to run tests multiple times
in the same process. useful for running tests from
a console.
"""
import py, sys
def pytest(argv=None):
if argv is None:
argv = []
try:
sys.argv[1:] = argv
py.cmdline.pytest()
except SystemExit:
pass
# we need to reset the global py.test.config object
py._com.comregistry = py._com.comregistry.__class__([])
py.test.config = py.test.config.__class__(
pluginmanager=py.test._PluginManager(py._com.comregistry))
#!/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