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

[svn r63020] allow to specify python executable in gatewayspecs, fix a few tests 

--HG--
branch : trunk
This commit is contained in:
hpk 2009-03-18 02:13:07 +01:00
parent fcaefb841b
commit 6f93561002
7 changed files with 60 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
py/execnet/gwmanage.py
View File

@ -24,11 +24,17 @@ from py.__.execnet.channel import RemoteError
NO_ENDMARKER_WANTED = object()
class GatewaySpec(object):
python = "python"
def __init__(self, spec, defaultjoinpath="pyexecnetcache"):
if spec == "popen" or spec.startswith("popen:"):
self.address = "popen"
self.joinpath = spec[len(self.address)+1:]
self.type = "popen"
parts = spec.split(":", 2)
self.type = self.address = parts.pop(0)
if parts:
self.python = parts.pop(0)
if parts:
self.joinpath = parts.pop(0)
else:
self.joinpath = ""
elif spec.startswith("socket:"):
parts = spec[7:].split(":", 2)
self.address = parts.pop(0)
@ -40,8 +46,9 @@ class GatewaySpec(object):
else:
if spec.startswith("ssh:"):
spec = spec[4:]
parts = spec.split(":", 1)
parts = spec.split(":", 2)
self.address = parts.pop(0)
self.python = parts and parts.pop(0) or "python"
self.joinpath = parts and parts.pop(0) or ""
self.type = "ssh"
if not self.joinpath and not self.inplacelocal():
@ -54,13 +61,13 @@ class GatewaySpec(object):
return "<GatewaySpec %s:%s>" % (self.address, self.joinpath)
__repr__ = __str__
def makegateway(self, python=None, waitclose=True):
def makegateway(self, waitclose=True):
if self.type == "popen":
gw = py.execnet.PopenGateway(python=python)
gw = py.execnet.PopenGateway(python=self.python)
elif self.type == "socket":
gw = py.execnet.SocketGateway(*self.address)
elif self.type == "ssh":
gw = py.execnet.SshGateway(self.address, remotepython=python)
gw = py.execnet.SshGateway(self.address, remotepython=self.python)
if self.joinpath:
channel = gw.remote_exec("""
import os

2
py/execnet/register.py
View File

@ -76,7 +76,7 @@ class PopenGateway(PopenCmdGateway):
""" instantiate a gateway to a subprocess
started with the given 'python' executable.
"""
if python is None:
if not python:
python = sys.executable
cmd = '%s -u -c "exec input()"' % python
super(PopenGateway, self).__init__(cmd)

5
py/execnet/testing/test_gateway.py
View File

@ -570,9 +570,8 @@ class TestSocketGateway(SocketGatewaySetup, BasicRemoteExecution):
class TestSshGateway(BasicRemoteExecution):
def setup_class(cls):
if py.test.config.option.sshhost is None:
py.test.skip("no known ssh target, use --sshhost to set one")
cls.gw = py.execnet.SshGateway(py.test.config.option.sshhost)
sshhost = py.test.config.getvalueorskip("sshhost")
cls.gw = py.execnet.SshGateway(sshhost)
def test_sshconfig_functional(self):
tmpdir = py.test.ensuretemp("test_sshconfig")

63
py/execnet/testing/test_gwmanage.py
View File

@ -18,29 +18,38 @@ class TestGatewaySpec:
[ssh:]spec:path SshGateway
* [SshGateway]
"""
def test_popen_nopath(self):
for joinpath in ('', ':abc', ':ab:cd', ':/x/y'):
spec = GatewaySpec("popen" + joinpath)
assert spec.address == "popen"
assert spec.joinpath == joinpath[1:]
assert spec.type == "popen"
spec2 = GatewaySpec("popen" + joinpath)
self._equality(spec, spec2)
def test_popen(self):
for python in ('', 'python2.4'):
for joinpath in ('', 'abc', 'ab:cd', '/x/y'):
s = ":".join(["popen", python, joinpath])
print s
spec = GatewaySpec(s)
assert spec.address == "popen"
assert spec.python == python
assert spec.joinpath == joinpath
assert spec.type == "popen"
spec2 = GatewaySpec("popen" + joinpath)
self._equality(spec, spec2)
def test_ssh(self):
for prefix in ('ssh:', ''): # ssh is default
for prefix in ('ssh', ''): # ssh is default
for hostpart in ('x.y', 'xyz@x.y'):
for joinpath in ('', ':abc', ':ab:cd', ':/tmp'):
specstring = prefix + hostpart + joinpath
spec = GatewaySpec(specstring)
assert spec.address == hostpart
if joinpath[1:]:
assert spec.joinpath == joinpath[1:]
else:
assert spec.joinpath == "pyexecnetcache"
assert spec.type == "ssh"
spec2 = GatewaySpec(specstring)
self._equality(spec, spec2)
for python in ('python', 'python2.5'):
for joinpath in ('', 'abc', 'ab:cd', '/tmp'):
specstring = ":".join([prefix, hostpart, python, joinpath])
if specstring[0] == ":":
specstring = specstring[1:]
print specstring
spec = GatewaySpec(specstring)
assert spec.address == hostpart
assert spec.python == python
if joinpath:
assert spec.joinpath == joinpath
else:
assert spec.joinpath == "pyexecnetcache"
assert spec.type == "ssh"
spec2 = GatewaySpec(specstring)
self._equality(spec, spec2)
def test_socket(self):
for hostpart in ('x.y', 'x', 'popen'):
@ -72,17 +81,17 @@ class TestGatewaySpecAPI:
gw.exit()
def test_popen_makegateway(self, testdir):
spec = GatewaySpec("popen:" + str(testdir.tmpdir))
spec = GatewaySpec("popen::" + str(testdir.tmpdir))
gw = spec.makegateway()
p = gw.remote_exec("import os; channel.send(os.getcwd())").receive()
assert spec.joinpath == p
gw.exit()
def test_popen_makegateway_python(self, testdir):
spec = GatewaySpec("popen")
gw = spec.makegateway(python=py.std.sys.executable)
spec = GatewaySpec("popen:%s" % py.std.sys.executable)
gw = spec.makegateway()
res = gw.remote_exec("import sys ; channel.send(sys.executable)").receive()
assert py.std.sys.executable == res
assert py.std.sys.executable == py.std.sys.executable
gw.exit()
def test_ssh(self):
@ -118,7 +127,7 @@ class TestGatewayManagerPopen:
assert not len(hm.gateways)
def test_hostmanager_rsync_popen_with_path(self, source, dest):
hm = GatewayManager(["popen:%s" %dest] * 1)
hm = GatewayManager(["popen::%s" %dest] * 1)
hm.makegateways()
source.ensure("dir1", "dir2", "hello")
l = []
@ -146,7 +155,7 @@ class TestGatewayManagerPopen:
def test_multi_chdir_popen_with_path(self, testdir):
import os
hm = GatewayManager(["popen:hello"] * 2)
hm = GatewayManager(["popen::hello"] * 2)
testdir.tmpdir.chdir()
hellopath = testdir.tmpdir.mkdir("hello")
hm.makegateways()
@ -253,7 +262,7 @@ class TestHRSync:
assert 'somedir' in basenames
def test_hrsync_one_host(self, source, dest):
spec = GatewaySpec("popen:%s" % dest)
spec = GatewaySpec("popen::%s" % dest)
gw = spec.makegateway()
finished = []
rsync = HostRSync(source)

2
py/io/testing/test_terminalwriter.py
View File

@ -18,7 +18,7 @@ def test_terminalwriter_defaultwidth_80():
py.magic.patch(terminalwriter, '_getdimensions', lambda: 0/0)
try:
tw = py.io.TerminalWriter()
assert tw.fullwidth == os.environ.get('COLUMNS', 80)-1
assert tw.fullwidth == int(os.environ.get('COLUMNS', 80)) -1
finally:
py.magic.revert(terminalwriter, '_getdimensions')

2
py/test/dsession/testing/test_functional_dsession.py
View File

@ -65,7 +65,7 @@ class TestAsyncFunctional:
p = subdir.join("test_one.py")
p.write("def test_5(): assert not __file__.startswith(%r)" % str(p))
result = testdir.runpytest("-d", "--rsyncdirs=%(subdir)s" % locals(),
"--gateways=popen:%(dest)s" % locals(), p)
"--gateways=popen::%(dest)s" % locals(), p)
assert result.ret == 0
result.stdout.fnmatch_lines([
"*1 passed*"

10
py/test/dsession/testing/test_hostmanage.py
View File

@ -37,7 +37,7 @@ class TestHostManager:
def test_hostmanager_rsync_roots_no_roots(self, source, dest):
source.ensure("dir1", "file1").write("hello")
config = py.test.config._reparse([source])
hm = HostManager(config, hosts=["popen:%s" % dest])
hm = HostManager(config, hosts=["popen::%s" % dest])
assert hm.config.topdir == source == config.topdir
hm.rsync_roots()
p, = hm.gwmanager.multi_exec("import os ; channel.send(os.getcwd())").receive_each()
@ -51,7 +51,7 @@ class TestHostManager:
dir2 = source.ensure("dir1", "dir2", dir=1)
dir2.ensure("hello")
hm = self.gethostmanager(source,
hosts = ["popen:%s" % dest],
hosts = ["popen::%s" % dest],
rsyncdirs = ['dir1']
)
assert hm.config.topdir == source
@ -64,7 +64,7 @@ class TestHostManager:
dir2 = source.ensure("dir1", "dir2", dir=1)
dir2.ensure("hello")
hm = self.gethostmanager(source,
hosts = ["popen:%s" % dest],
hosts = ["popen::%s" % dest],
rsyncdirs = [str(source)]
)
assert hm.config.topdir == source
@ -84,7 +84,7 @@ class TestHostManager:
"""))
session = py.test.config._reparse([source]).initsession()
hm = HostManager(session.config,
hosts=["popen:" + str(dest)])
hosts=["popen::" + str(dest)])
hm.rsync_roots()
assert dest.join("dir2").check()
assert not dest.join("dir1").check()
@ -101,7 +101,7 @@ class TestHostManager:
"""))
session = py.test.config._reparse([source]).initsession()
hm = HostManager(session.config,
hosts=["popen:" + str(dest)])
hosts=["popen::" + str(dest)])
hm.rsync_roots()
assert dest.join("dir1").check()
assert not dest.join("dir1", "dir2").check()