floraachy
/
pytest2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

* moving execnet tests to funcarg-style, some cleanup 

* slight refinement to FAQ license topic

--HG--
branch : trunk
This commit is contained in:
holger krekel 2009-09-08 10:10:36 +02:00
parent f9eadc6440
commit b70c7a209d
3 changed files with 186 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
doc/faq.txt
View File

@ -51,7 +51,7 @@ have no counterpart in nose_.
Why did you choose a GPL-style license? Why did you choose a GPL-style license?
---------------------------------------- ----------------------------------------
Older versions of the py lib and (up until 1.0.x) Older versions of the py lib and py.test (up until 1.0.x)
were licensed under the MIT license. Starting were licensed under the MIT license. Starting
with the 1.1 series Holger Krekel - being the main maintainer with the 1.1 series Holger Krekel - being the main maintainer
and developer since several years - decided to go for and developer since several years - decided to go for
@ -66,23 +66,25 @@ a GPL-style license mainly for these reasons:
Developers want to co-operate no matter what context they Developers want to co-operate no matter what context they
are in, commercial, free, whatever. BSD-licenses sound like are in, commercial, free, whatever. BSD-licenses sound like
a fit because they minimize the need for checking for a fit because they minimize the need for checking for
constraints from the company or legal department. constraints from the company or legal department. They allow
to use and modify software for whatever purpose.
Developers wanting to produce free software for a living also However, developers wanting to produce free software for a living
want to connect to a sustainable revenue system, however. When often need to connect to a sustainable revenue system. When
releasing software for public use they want to seek means, releasing software for public use they seek means, some security
some security on getting something back: Contributions, on getting something back: Contributions, recognition or money.
recognition or money. The GPL license tries to foster a The GPL license tries to foster a universe of free software and
universe of free software and force proprietary players force proprietary players to contribute back.
to contribute back.
Choosing the Lesser GPL kind of strikes a balance - it allows The py lib choose the Lesser GPL. It strikes a balance because it
the code to interact in proprietary contexts but increases allows the code to interact in proprietary contexts and increases
likelyness of flow backs. Practically it all does not make likelyness of flow backs.
much of a difference. Anyway, if you do have actual practical
issues regarding the license please just get in contact. If you do have or get actual practical issues regarding
licensing please get in contact_.
.. _fsf: http://www.fsf.org .. _fsf: http://www.fsf.org
.. _contact: contact.html
What's all this "magic" with py.test? What's all this "magic" with py.test?
---------------------------------------- ----------------------------------------

2
py/execnet/gateway.py
View File

@ -1,6 +1,5 @@
import sys, os, inspect, socket, atexit, weakref import sys, os, inspect, socket, atexit, weakref
import py import py
from subprocess import Popen, PIPE
from py.__.execnet.gateway_base import BaseGateway, Message, Popen2IO, SocketIO from py.__.execnet.gateway_base import BaseGateway, Message, Popen2IO, SocketIO
from py.__.execnet.gateway_base import ExecnetAPI from py.__.execnet.gateway_base import ExecnetAPI
@ -196,6 +195,7 @@ channel.send(dict(
class PopenCmdGateway(InitiatingGateway): class PopenCmdGateway(InitiatingGateway):
def __init__(self, args): def __init__(self, args):
from subprocess import Popen, PIPE
self._popen = p = Popen(args, stdin=PIPE, stdout=PIPE) self._popen = p = Popen(args, stdin=PIPE, stdout=PIPE)
io = Popen2IO(p.stdin, p.stdout) io = Popen2IO(p.stdin, p.stdout)
super(PopenCmdGateway, self).__init__(io=io) super(PopenCmdGateway, self).__init__(io=io)

360
testing/execnet/test_gateway.py
View File

@ -1,34 +1,27 @@
from __future__ import generators """
mostly functional tests of gateways.
"""
import os, sys, time import os, sys, time
import py import py
from py.__.execnet import gateway_base, gateway from py.__.execnet import gateway_base, gateway
queue = py.builtin._tryimport('queue', 'Queue') queue = py.builtin._tryimport('queue', 'Queue')
pytest_plugins = "pytester"
TESTTIMEOUT = 10.0 # seconds TESTTIMEOUT = 10.0 # seconds
class TestBasicRemoteExecution:
def test_correct_setup(self, gw):
assert gw._receiverthread.isAlive()
class PopenGatewayTestSetup: def test_repr_doesnt_crash(self, gw):
def setup_class(cls): assert isinstance(repr(gw), str)
cls.gw = py.execnet.PopenGateway()
#def teardown_class(cls): def test_attribute__name__(self, gw):
# cls.gw.exit() channel = gw.remote_exec("channel.send(__name__)")
class BasicRemoteExecution:
def test_correct_setup(self):
assert self.gw._receiverthread.isAlive()
def test_repr_doesnt_crash(self):
assert isinstance(repr(self.gw), str)
def test_attribute__name__(self):
channel = self.gw.remote_exec("channel.send(__name__)")
name = channel.receive() name = channel.receive()
assert name == "__channelexec__" assert name == "__channelexec__"
def test_correct_setup_no_py(self): def test_correct_setup_no_py(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
import sys import sys
channel.send(list(sys.modules)) channel.send(list(sys.modules))
""") """)
@ -36,25 +29,25 @@ class BasicRemoteExecution:
assert 'py' not in remotemodules, ( assert 'py' not in remotemodules, (
"py should not be imported on remote side") "py should not be imported on remote side")
def test_remote_exec_waitclose(self): def test_remote_exec_waitclose(self, gw):
channel = self.gw.remote_exec('pass') channel = gw.remote_exec('pass')
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
def test_remote_exec_waitclose_2(self): def test_remote_exec_waitclose_2(self, gw):
channel = self.gw.remote_exec('def gccycle(): pass') channel = gw.remote_exec('def gccycle(): pass')
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
def test_remote_exec_waitclose_noarg(self): def test_remote_exec_waitclose_noarg(self, gw):
channel = self.gw.remote_exec('pass') channel = gw.remote_exec('pass')
channel.waitclose() channel.waitclose()
def test_remote_exec_error_after_close(self): def test_remote_exec_error_after_close(self, gw):
channel = self.gw.remote_exec('pass') channel = gw.remote_exec('pass')
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
py.test.raises(IOError, channel.send, 0) py.test.raises(IOError, channel.send, 0)
def test_remote_exec_channel_anonymous(self): def test_remote_exec_channel_anonymous(self, gw):
channel = self.gw.remote_exec(''' channel = gw.remote_exec('''
obj = channel.receive() obj = channel.receive()
channel.send(obj) channel.send(obj)
''') ''')
@ -62,37 +55,38 @@ class BasicRemoteExecution:
result = channel.receive() result = channel.receive()
assert result == 42 assert result == 42
def test_channel_close_and_then_receive_error(self): class TestChannelBasicBehaviour:
channel = self.gw.remote_exec('raise ValueError') def test_channel_close_and_then_receive_error(self, gw):
channel = gw.remote_exec('raise ValueError')
py.test.raises(channel.RemoteError, channel.receive) py.test.raises(channel.RemoteError, channel.receive)
def test_channel_finish_and_then_EOFError(self): def test_channel_finish_and_then_EOFError(self, gw):
channel = self.gw.remote_exec('channel.send(42)') channel = gw.remote_exec('channel.send(42)')
x = channel.receive() x = channel.receive()
assert x == 42 assert x == 42
py.test.raises(EOFError, channel.receive) py.test.raises(EOFError, channel.receive)
py.test.raises(EOFError, channel.receive) py.test.raises(EOFError, channel.receive)
py.test.raises(EOFError, channel.receive) py.test.raises(EOFError, channel.receive)
def test_channel_close_and_then_receive_error_multiple(self): def test_channel_close_and_then_receive_error_multiple(self, gw):
channel = self.gw.remote_exec('channel.send(42) ; raise ValueError') channel = gw.remote_exec('channel.send(42) ; raise ValueError')
x = channel.receive() x = channel.receive()
assert x == 42 assert x == 42
py.test.raises(channel.RemoteError, channel.receive) py.test.raises(channel.RemoteError, channel.receive)
def test_channel__local_close(self): def test_channel__local_close(self, gw):
channel = self.gw._channelfactory.new() channel = gw._channelfactory.new()
self.gw._channelfactory._local_close(channel.id) gw._channelfactory._local_close(channel.id)
channel.waitclose(0.1) channel.waitclose(0.1)
def test_channel__local_close_error(self): def test_channel__local_close_error(self, gw):
channel = self.gw._channelfactory.new() channel = gw._channelfactory.new()
self.gw._channelfactory._local_close(channel.id, gw._channelfactory._local_close(channel.id,
channel.RemoteError("error")) channel.RemoteError("error"))
py.test.raises(channel.RemoteError, channel.waitclose, 0.01) py.test.raises(channel.RemoteError, channel.waitclose, 0.01)
def test_channel_error_reporting(self): def test_channel_error_reporting(self, gw):
channel = self.gw.remote_exec('def foo():\n return foobar()\nfoo()\n') channel = gw.remote_exec('def foo():\n return foobar()\nfoo()\n')
try: try:
channel.receive() channel.receive()
except channel.RemoteError: except channel.RemoteError:
@ -103,9 +97,9 @@ class BasicRemoteExecution:
else: else:
py.test.fail('No exception raised') py.test.fail('No exception raised')
def test_channel_syntax_error(self): def test_channel_syntax_error(self, gw):
# missing colon # missing colon
channel = self.gw.remote_exec('def foo()\n return 1\nfoo()\n') channel = gw.remote_exec('def foo()\n return 1\nfoo()\n')
try: try:
channel.receive() channel.receive()
except channel.RemoteError: except channel.RemoteError:
@ -113,16 +107,16 @@ class BasicRemoteExecution:
assert str(e).startswith('Traceback (most recent call last):') assert str(e).startswith('Traceback (most recent call last):')
assert str(e).find('SyntaxError') > -1 assert str(e).find('SyntaxError') > -1
def test_channel_iter(self): def test_channel_iter(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
for x in range(3): for x in range(3):
channel.send(x) channel.send(x)
""") """)
l = list(channel) l = list(channel)
assert l == [0, 1, 2] assert l == [0, 1, 2]
def test_channel_passing_over_channel(self): def test_channel_passing_over_channel(self, gw):
channel = self.gw.remote_exec(''' channel = gw.remote_exec('''
c = channel.gateway.newchannel() c = channel.gateway.newchannel()
channel.send(c) channel.send(c)
c.send(42) c.send(42)
@ -133,17 +127,17 @@ class BasicRemoteExecution:
# check that the both sides previous channels are really gone # check that the both sides previous channels are really gone
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
#assert c.id not in self.gw._channelfactory #assert c.id not in gw._channelfactory
newchan = self.gw.remote_exec(''' newchan = gw.remote_exec('''
assert %d not in channel.gateway._channelfactory._channels assert %d not in channel.gateway._channelfactory._channels
''' % (channel.id)) ''' % (channel.id))
newchan.waitclose(TESTTIMEOUT) newchan.waitclose(TESTTIMEOUT)
assert channel.id not in self.gw._channelfactory._channels assert channel.id not in gw._channelfactory._channels
def test_channel_receiver_callback(self): def test_channel_receiver_callback(self, gw):
l = [] l = []
#channel = self.gw.newchannel(receiver=l.append) #channel = gw.newchannel(receiver=l.append)
channel = self.gw.remote_exec(source=''' channel = gw.remote_exec(source='''
channel.send(42) channel.send(42)
channel.send(13) channel.send(13)
channel.send(channel.gateway.newchannel()) channel.send(channel.gateway.newchannel())
@ -155,9 +149,9 @@ class BasicRemoteExecution:
assert l[:2] == [42,13] assert l[:2] == [42,13]
assert isinstance(l[2], channel.__class__) assert isinstance(l[2], channel.__class__)
def test_channel_callback_after_receive(self): def test_channel_callback_after_receive(self, gw):
l = [] l = []
channel = self.gw.remote_exec(source=''' channel = gw.remote_exec(source='''
channel.send(42) channel.send(42)
channel.send(13) channel.send(13)
channel.send(channel.gateway.newchannel()) channel.send(channel.gateway.newchannel())
@ -171,25 +165,25 @@ class BasicRemoteExecution:
assert l[0] == 13 assert l[0] == 13
assert isinstance(l[1], channel.__class__) assert isinstance(l[1], channel.__class__)
def test_waiting_for_callbacks(self): def test_waiting_for_callbacks(self, gw):
l = [] l = []
def callback(msg): def callback(msg):
import time; time.sleep(0.2) import time; time.sleep(0.2)
l.append(msg) l.append(msg)
channel = self.gw.remote_exec(source=''' channel = gw.remote_exec(source='''
channel.send(42) channel.send(42)
''') ''')
channel.setcallback(callback) channel.setcallback(callback)
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
assert l == [42] assert l == [42]
def test_channel_callback_stays_active(self): def test_channel_callback_stays_active(self, gw):
self.check_channel_callback_stays_active(earlyfree=True) self.check_channel_callback_stays_active(gw, earlyfree=True)
def check_channel_callback_stays_active(self, earlyfree=True): def check_channel_callback_stays_active(self, gw, earlyfree=True):
# with 'earlyfree==True', this tests the "sendonly" channel state. # with 'earlyfree==True', this tests the "sendonly" channel state.
l = [] l = []
channel = self.gw.remote_exec(source=''' channel = gw.remote_exec(source='''
try: try:
import thread import thread
except ImportError: except ImportError:
@ -203,7 +197,7 @@ class BasicRemoteExecution:
thread.start_new_thread(producer, (channel2,)) thread.start_new_thread(producer, (channel2,))
del channel2 del channel2
''') ''')
subchannel = self.gw.newchannel() subchannel = gw.newchannel()
subchannel.setcallback(l.append) subchannel.setcallback(l.append)
channel.send(subchannel) channel.send(subchannel)
if earlyfree: if earlyfree:
@ -220,13 +214,14 @@ class BasicRemoteExecution:
assert l == [0, 100, 200, 300, 400] assert l == [0, 100, 200, 300, 400]
return subchannel return subchannel
def test_channel_callback_remote_freed(self): def test_channel_callback_remote_freed(self, gw):
channel = self.check_channel_callback_stays_active(earlyfree=False) channel = self.check_channel_callback_stays_active(gw, earlyfree=False)
channel.waitclose(TESTTIMEOUT) # freed automatically at the end of producer() # freed automatically at the end of producer()
channel.waitclose(TESTTIMEOUT)
def test_channel_endmarker_callback(self): def test_channel_endmarker_callback(self, gw):
l = [] l = []
channel = self.gw.remote_exec(source=''' channel = gw.remote_exec(source='''
channel.send(42) channel.send(42)
channel.send(13) channel.send(13)
channel.send(channel.gateway.newchannel()) channel.send(channel.gateway.newchannel())
@ -239,9 +234,9 @@ class BasicRemoteExecution:
assert isinstance(l[2], channel.__class__) assert isinstance(l[2], channel.__class__)
assert l[3] == 999 assert l[3] == 999
def test_channel_endmarker_callback_error(self): def test_channel_endmarker_callback_error(self, gw):
q = queue.Queue() q = queue.Queue()
channel = self.gw.remote_exec(source=''' channel = gw.remote_exec(source='''
raise ValueError() raise ValueError()
''') ''')
channel.setcallback(q.put, endmarker=999) channel.setcallback(q.put, endmarker=999)
@ -252,20 +247,20 @@ class BasicRemoteExecution:
assert str(err).find("ValueError") != -1 assert str(err).find("ValueError") != -1
@py.test.mark.xfail @py.test.mark.xfail
def test_remote_redirect_stdout(self): def test_remote_redirect_stdout(self, gw):
out = py.io.TextIO() out = py.io.TextIO()
handle = self.gw._remote_redirect(stdout=out) handle = gw._remote_redirect(stdout=out)
c = self.gw.remote_exec("print 42") c = gw.remote_exec("print 42")
c.waitclose(TESTTIMEOUT) c.waitclose(TESTTIMEOUT)
handle.close() handle.close()
s = out.getvalue() s = out.getvalue()
assert s.strip() == "42" assert s.strip() == "42"
@py.test.mark.xfail @py.test.mark.xfail
def test_remote_exec_redirect_multi(self): def test_remote_exec_redirect_multi(self, gw):
num = 3 num = 3
l = [[] for x in range(num)] l = [[] for x in range(num)]
channels = [self.gw.remote_exec("print %d" % i, channels = [gw.remote_exec("print %d" % i,
stdout=l[i].append) stdout=l[i].append)
for i in range(num)] for i in range(num)]
for x in channels: for x in channels:
@ -277,8 +272,9 @@ class BasicRemoteExecution:
s = subl[0] s = subl[0]
assert s.strip() == str(i) assert s.strip() == str(i)
def test_channel_file_write(self): class TestChannelFile:
channel = self.gw.remote_exec(""" def test_channel_file_write(self, gw):
channel = gw.remote_exec("""
f = channel.makefile() f = channel.makefile()
f.write("hello world\\n") f.write("hello world\\n")
f.close() f.close()
@ -289,14 +285,14 @@ class BasicRemoteExecution:
second = channel.receive() second = channel.receive()
assert second == 42 assert second == 42
def test_channel_file_write_error(self): def test_channel_file_write_error(self, gw):
channel = self.gw.remote_exec("pass") channel = gw.remote_exec("pass")
f = channel.makefile() f = channel.makefile()
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
py.test.raises(IOError, f.write, 'hello') py.test.raises(IOError, f.write, 'hello')
def test_channel_file_proxyclose(self): def test_channel_file_proxyclose(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
f = channel.makefile(proxyclose=True) f = channel.makefile(proxyclose=True)
f.write("hello world") f.write("hello world")
f.close() f.close()
@ -306,8 +302,8 @@ class BasicRemoteExecution:
assert first.strip() == 'hello world' assert first.strip() == 'hello world'
py.test.raises(EOFError, channel.receive) py.test.raises(EOFError, channel.receive)
def test_channel_file_read(self): def test_channel_file_read(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
f = channel.makefile(mode='r') f = channel.makefile(mode='r')
s = f.read(2) s = f.read(2)
channel.send(s) channel.send(s)
@ -320,16 +316,16 @@ class BasicRemoteExecution:
assert s1 == "xy" assert s1 == "xy"
assert s2 == "abcde" assert s2 == "abcde"
def test_channel_file_read_empty(self): def test_channel_file_read_empty(self, gw):
channel = self.gw.remote_exec("pass") channel = gw.remote_exec("pass")
f = channel.makefile(mode="r") f = channel.makefile(mode="r")
s = f.read(3) s = f.read(3)
assert s == "" assert s == ""
s = f.read(5) s = f.read(5)
assert s == "" assert s == ""
def test_channel_file_readline_remote(self): def test_channel_file_readline_remote(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
channel.send('123\\n45') channel.send('123\\n45')
""") """)
channel.waitclose(TESTTIMEOUT) channel.waitclose(TESTTIMEOUT)
@ -339,12 +335,12 @@ class BasicRemoteExecution:
s = f.readline() s = f.readline()
assert s == "45" assert s == "45"
def test_channel_makefile_incompatmode(self): def test_channel_makefile_incompatmode(self, gw):
channel = self.gw.newchannel() channel = gw.newchannel()
py.test.raises(ValueError, 'channel.makefile("rw")') py.test.raises(ValueError, 'channel.makefile("rw")')
def test_confusion_from_os_write_stdout(self): def test_confusion_from_os_write_stdout(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
import os import os
os.write(1, 'confusion!'.encode('ascii')) os.write(1, 'confusion!'.encode('ascii'))
channel.send(channel.receive() * 6) channel.send(channel.receive() * 6)
@ -357,8 +353,8 @@ class BasicRemoteExecution:
res = channel.receive() res = channel.receive()
assert res == 42 assert res == 42
def test_confusion_from_os_write_stderr(self): def test_confusion_from_os_write_stderr(self, gw):
channel = self.gw.remote_exec(""" channel = gw.remote_exec("""
import os import os
os.write(2, 'test'.encode('ascii')) os.write(2, 'test'.encode('ascii'))
channel.send(channel.receive() * 6) channel.send(channel.receive() * 6)
@ -371,53 +367,44 @@ class BasicRemoteExecution:
res = channel.receive() res = channel.receive()
assert res == 42 assert res == 42
def test__rinfo(self): def test__rinfo(self, gw):
rinfo = self.gw._rinfo() rinfo = gw._rinfo()
assert rinfo.executable assert rinfo.executable
assert rinfo.cwd assert rinfo.cwd
assert rinfo.version_info assert rinfo.version_info
s = repr(rinfo) s = repr(rinfo)
old = self.gw.remote_exec(""" old = gw.remote_exec("""
import os.path import os.path
cwd = os.getcwd() cwd = os.getcwd()
channel.send(os.path.basename(cwd)) channel.send(os.path.basename(cwd))
os.chdir('..') os.chdir('..')
""").receive() """).receive()
try: try:
rinfo2 = self.gw._rinfo() rinfo2 = gw._rinfo()
assert rinfo2.cwd == rinfo.cwd assert rinfo2.cwd == rinfo.cwd
rinfo3 = self.gw._rinfo(update=True) rinfo3 = gw._rinfo(update=True)
assert rinfo3.cwd != rinfo2.cwd assert rinfo3.cwd != rinfo2.cwd
finally: finally:
self.gw._cache_rinfo = rinfo gw._cache_rinfo = rinfo
self.gw.remote_exec("import os ; os.chdir(%r)" % old).waitclose() gw.remote_exec("import os ; os.chdir(%r)" % old).waitclose()
class BasicCmdbasedRemoteExecution(BasicRemoteExecution): def test_join_blocked_execution_gateway():
def test_cmdattr(self): gateway = py.execnet.PopenGateway()
assert hasattr(self.gw, '_cmd') channel = gateway.remote_exec("""
time.sleep(5.0)
""")
def doit():
gateway.exit()
gateway.join(joinexec=True)
return 17
#class TestBlockingIssues: pool = py._thread.WorkerPool()
# def test_join_blocked_execution_gateway(self): reply = pool.dispatch(doit)
# gateway = py.execnet.PopenGateway() x = reply.get(timeout=1.0)
# channel = gateway.remote_exec(""" assert x == 17
# time.sleep(5.0)
# """)
# def doit():
# gateway.exit()
# gateway.join(joinexec=True)
# return 17
#
# pool = py._thread.WorkerPool()
# reply = pool.dispatch(doit)
# x = reply.get(timeout=1.0)
# assert x == 17
class TestPopenGateway(PopenGatewayTestSetup, BasicRemoteExecution): class TestPopenGateway:
def test_rinfo_popen(self): gwtype = 'popen'
rinfo = self.gw._rinfo()
assert rinfo.executable == py.std.sys.executable
assert rinfo.cwd == py.std.os.getcwd()
assert rinfo.version_info == py.std.sys.version_info
def test_chdir_separation(self, tmpdir): def test_chdir_separation(self, tmpdir):
old = tmpdir.chdir() old = tmpdir.chdir()
@ -428,7 +415,7 @@ class TestPopenGateway(PopenGatewayTestSetup, BasicRemoteExecution):
c = gw.remote_exec("import os ; channel.send(os.getcwd())") c = gw.remote_exec("import os ; channel.send(os.getcwd())")
x = c.receive() x = c.receive()
assert x == str(waschangedir) assert x == str(waschangedir)
def test_many_popen(self): def test_many_popen(self):
num = 4 num = 4
l = [] l = []
@ -453,6 +440,21 @@ class TestPopenGateway(PopenGatewayTestSetup, BasicRemoteExecution):
channel = channels.pop() channel = channels.pop()
ret = channel.receive() ret = channel.receive()
assert ret == 42 assert ret == 42
def test_rinfo_popen(self, gw):
rinfo = gw._rinfo()
assert rinfo.executable == py.std.sys.executable
assert rinfo.cwd == py.std.os.getcwd()
assert rinfo.version_info == py.std.sys.version_info
def test_gateway_init_event(self, _pytest):
rec = _pytest.gethookrecorder(gateway_base.ExecnetAPI)
gw = py.execnet.PopenGateway()
call = rec.popcall("pyexecnet_gateway_init")
assert call.gateway == gw
gw.exit()
call = rec.popcall("pyexecnet_gateway_exit")
assert call.gateway == gw
@py.test.mark.xfail # "fix needed: dying remote process does not cause waitclose() to fail" @py.test.mark.xfail # "fix needed: dying remote process does not cause waitclose() to fail"
def test_waitclose_on_remote_killed(self): def test_waitclose_on_remote_killed(self):
@ -491,75 +493,51 @@ def test_endmarker_delivery_on_remote_killterm():
assert "15" in str(err) assert "15" in str(err)
class SocketGatewaySetup: def test_socket_gw_host_not_found(gw):
def setup_class(cls): py.test.raises(py.execnet.HostNotFound,
# open a gateway to a fresh child process 'py.execnet.SocketGateway("qowieuqowe", 9000)'
cls.proxygw = py.execnet.PopenGateway() )
cls.gw = py.execnet.SocketGateway.new_remote(cls.proxygw,
("127.0.0.1", 0) class TestSshPopenGateway:
) gwtype = "ssh"
def test_sshconfig_config_parsing(self, monkeypatch):
import subprocess
l = []
monkeypatch.setattr(subprocess, 'Popen',
lambda *args, **kwargs: l.append(args[0]))
py.test.raises(AttributeError,
"""py.execnet.SshGateway("xyz", ssh_config='qwe')""")
assert len(l) == 1
popen_args = l[0]
i = popen_args.index('-F')
assert popen_args[i+1] == "qwe"
def test_sshaddress(self, gw, specssh):
assert gw.remoteaddress == specssh.ssh
def test_host_not_found(self): def test_host_not_found(self):
py.test.raises(py.execnet.HostNotFound,
'py.execnet.SocketGateway("qowieuqowe", 9000)'
)
## def teardown_class(cls):
## cls.gw.exit()
## cls.proxygw.exit()
class TestSocketGateway(SocketGatewaySetup, BasicRemoteExecution):
pass
class TestSshGateway(BasicRemoteExecution):
def setup_class(cls):
from conftest import getspecssh
cls.sshhost = getspecssh().ssh
cls.gw = py.execnet.SshGateway(cls.sshhost)
def test_sshconfig_functional(self, tmpdir):
ssh_config = tmpdir.join("ssh_config")
ssh_config.write(
"Host alias123\n"
" HostName %s\n" % self.sshhost)
gw = py.execnet.SshGateway("alias123", ssh_config=ssh_config)
pid = gw.remote_exec("import os ; channel.send(os.getpid())").receive()
gw.exit()
def test_sshaddress(self):
assert self.gw.remoteaddress == self.sshhost
def test_connexion_failes_on_non_existing_hosts(self):
py.test.raises(py.execnet.HostNotFound, py.test.raises(py.execnet.HostNotFound,
"py.execnet.SshGateway('nowhere.codespeak.net')") "py.execnet.SshGateway('nowhere.codespeak.net')")
def test_threads(): class TestThreads:
gw = py.execnet.PopenGateway() def test_threads(self):
gw.remote_init_threads(3)
c1 = gw.remote_exec("channel.send(channel.receive())")
c2 = gw.remote_exec("channel.send(channel.receive())")
c2.send(1)
res = c2.receive()
assert res == 1
c1.send(42)
res = c1.receive()
assert res == 42
gw.exit()
def test_threads_twice():
gw = py.execnet.PopenGateway()
gw.remote_init_threads(3)
py.test.raises(IOError, gw.remote_init_threads, 3)
gw.exit()
class TestExecnetEvents:
def test_popengateway(self, _pytest):
rec = _pytest.gethookrecorder(gateway_base.ExecnetAPI)
gw = py.execnet.PopenGateway() gw = py.execnet.PopenGateway()
call = rec.popcall("pyexecnet_gateway_init") gw.remote_init_threads(3)
assert call.gateway == gw c1 = gw.remote_exec("channel.send(channel.receive())")
gw.exit() c2 = gw.remote_exec("channel.send(channel.receive())")
call = rec.popcall("pyexecnet_gateway_exit") c2.send(1)
assert call.gateway == gw res = c2.receive()
assert res == 1
c1.send(42)
res = c1.receive()
assert res == 42
def test_threads_twice(self):
gw = py.execnet.PopenGateway()
gw.remote_init_threads(3)
py.test.raises(IOError, gw.remote_init_threads, 3)
def test_nodebug(): def test_nodebug():
from py.__.execnet import gateway_base from py.__.execnet import gateway_base