""" |
|
Put this file as 'conftest.py' somewhere upwards from py-trunk, |
modify the "socketserveradr" below to point to a windows/linux |
host running "py/execnet/script/loop_socketserver.py" |
and invoke e.g. from linux: |
|
py.test --session=MySession some_path_to_what_you_want_to_test |
|
This should ad-hoc distribute the running of tests to |
the remote machine (including rsyncing your WC). |
|
""" |
import py |
from py.__.test.terminal.remote import RemoteTerminalSession |
|
import os |
|
class MyRSync(py.execnet.RSync): |
def filter(self, path): |
if path.endswith('.pyc') or path.endswith('~'): |
return False |
dir, base = os.path.split(path) |
|
|
|
|
return True |
|
class MySession(RemoteTerminalSession): |
socketserveradr = ('10.9.2.62', 8888) |
socketserveradr = ('10.9.4.148', 8888) |
|
def _initslavegateway(self): |
print "MASTER: initializing remote socket gateway" |
gw = py.execnet.SocketGateway(*self.socketserveradr) |
pkgname = 'py' |
channel = gw.remote_exec(""" |
import os |
topdir = os.path.join(os.environ['HOMEPATH'], 'pytestcache') |
pkgdir = os.path.join(topdir, %r) |
channel.send((topdir, pkgdir)) |
""" % (pkgname,)) |
remotetopdir, remotepkgdir = channel.receive() |
sendpath = py.path.local(py.__file__).dirpath() |
rsync = MyRSync(sendpath) |
rsync.add_target(gw, remotepkgdir, delete=True) |
rsync.send() |
channel = gw.remote_exec(""" |
import os, sys |
path = %r # os.path.abspath |
sys.path.insert(0, path) |
os.chdir(path) |
import py |
channel.send((path, py.__file__)) |
""" % remotetopdir) |
topdir, remotepypath = channel.receive() |
assert topdir == remotetopdir, (topdir, remotetopdir) |
assert remotepypath.startswith(topdir), (remotepypath, topdir) |
|
return gw, topdir |
|
dist_hosts = ['localhost', 'cobra', 'cobra'] |
|