[py-svn] r63040 - in py/trunk/py: execnet execnet/testing test test/plugin
hpk at codespeak.net
hpk at codespeak.net
Wed Mar 18 16:51:57 CET 2009
Author: hpk
Date: Wed Mar 18 16:51:55 2009
New Revision: 63040
Added:
py/trunk/py/execnet/testing/test_event.py
py/trunk/py/test/plugin/pytest_execnetcleanup.py
Modified:
py/trunk/py/execnet/gateway.py
py/trunk/py/execnet/register.py
py/trunk/py/execnet/testing/test_gateway.py
py/trunk/py/test/defaultconftest.py
py/trunk/py/test/plugin/pytest_plugintester.py
py/trunk/py/test/plugin/pytest_pytester.py
py/trunk/py/test/pytestplugin.py
Log:
try harder to record and auto-exit gateways after test runs
Modified: py/trunk/py/execnet/gateway.py
==============================================================================
--- py/trunk/py/execnet/gateway.py (original)
+++ py/trunk/py/execnet/gateway.py Wed Mar 18 16:51:55 2009
@@ -330,6 +330,7 @@
self._cleanup.unregister(self)
self._stopexec()
self._stopsend()
+ py._com.pyplugins.notify("gateway_exit", self)
def _stopsend(self):
self._send(None)
Modified: py/trunk/py/execnet/register.py
==============================================================================
--- py/trunk/py/execnet/register.py (original)
+++ py/trunk/py/execnet/register.py Wed Mar 18 16:51:55 2009
@@ -41,6 +41,7 @@
super(InstallableGateway, self).__init__(io=io, _startcount=1)
# XXX we dissallow execution form the other side
self._initreceive(requestqueue=False)
+ py._com.pyplugins.notify("gateway_init", self)
def _remote_bootstrap_gateway(self, io, extra=''):
""" return Gateway with a asynchronously remotely
Added: py/trunk/py/execnet/testing/test_event.py
==============================================================================
--- (empty file)
+++ py/trunk/py/execnet/testing/test_event.py Wed Mar 18 16:51:55 2009
@@ -0,0 +1,11 @@
+import py
+pytest_plugins = "pytester"
+
+class TestExecnetEvents:
+ def test_popengateway(self, eventrecorder):
+ gw = py.execnet.PopenGateway()
+ event = eventrecorder.popevent("gateway_init")
+ assert event.args[0] == gw
+ gw.exit()
+ event = eventrecorder.popevent("gateway_exit")
+ assert event.args[0] == gw
Modified: py/trunk/py/execnet/testing/test_gateway.py
==============================================================================
--- py/trunk/py/execnet/testing/test_gateway.py (original)
+++ py/trunk/py/execnet/testing/test_gateway.py Wed Mar 18 16:51:55 2009
@@ -571,6 +571,9 @@
class TestSshGateway(BasicRemoteExecution):
def setup_class(cls):
sshhost = py.test.config.getvalueorskip("sshhost")
+ if sshhost.find(":") != -1:
+ sshhost = sshhost.split(":")[0]
+ cls.sshhost = sshhost
cls.gw = py.execnet.SshGateway(sshhost)
def test_sshconfig_functional(self):
@@ -578,7 +581,7 @@
ssh_config = tmpdir.join("ssh_config")
ssh_config.write(
"Host alias123\n"
- " HostName %s\n" % (py.test.config.option.sshhost,))
+ " HostName %s\n" % self.sshhost)
gw = py.execnet.SshGateway("alias123", ssh_config=ssh_config)
assert gw._cmd.find("-F") != -1
assert gw._cmd.find(str(ssh_config)) != -1
@@ -586,7 +589,7 @@
gw.exit()
def test_sshaddress(self):
- assert self.gw.remoteaddress == py.test.config.option.sshhost
+ assert self.gw.remoteaddress == self.sshhost
@py.test.mark.xfail("XXX ssh-gateway error handling")
def test_connexion_failes_on_non_existing_hosts(self):
Modified: py/trunk/py/test/defaultconftest.py
==============================================================================
--- py/trunk/py/test/defaultconftest.py (original)
+++ py/trunk/py/test/defaultconftest.py Wed Mar 18 16:51:55 2009
@@ -11,7 +11,7 @@
conf_iocapture = "fd" # overridable from conftest.py
# XXX resultlog should go, pypy's nightrun depends on it
-pytest_plugins = "default terminal xfail tmpdir resultlog monkeypatch".split()
+pytest_plugins = "default terminal xfail tmpdir execnetcleanup resultlog monkeypatch".split()
# ===================================================
# Distributed testing specific options
Added: py/trunk/py/test/plugin/pytest_execnetcleanup.py
==============================================================================
--- (empty file)
+++ py/trunk/py/test/plugin/pytest_execnetcleanup.py Wed Mar 18 16:51:55 2009
@@ -0,0 +1,53 @@
+import py
+
+class ExecnetcleanupPlugin:
+ _gateways = None
+ _debug = None
+
+ def pytest_configure(self, config):
+ self._debug = config.option.debug
+
+ def trace(self, msg, *args):
+ if self._debug:
+ print "[execnetcleanup %0x] %s %s" %(id(self), msg, args)
+
+ def pyevent_gateway_init(self, gateway):
+ self.trace("init", gateway)
+ if self._gateways is not None:
+ self._gateways.append(gateway)
+
+ def pyevent_gateway_exit(self, gateway):
+ self.trace("exit", gateway)
+ if self._gateways is not None:
+ self._gateways.remove(gateway)
+
+ def pyevent_testrunstart(self, event):
+ self.trace("testrunstart", event)
+ self._gateways = []
+
+ def pyevent_testrunfinish(self, event):
+ self.trace("testrunfinish", event)
+ l = []
+ for gw in self._gateways:
+ gw.exit()
+ l.append(gw)
+ for gw in l:
+ gw.join()
+
+def test_generic(plugintester):
+ plugintester.apicheck(ExecnetcleanupPlugin)
+
+ at py.test.mark.xfail("clarify plugin registration/unregistration")
+def test_execnetplugin(testdir):
+ p = ExecnetcleanupPlugin()
+ testdir.plugins.append(p)
+ testdir.inline_runsource("""
+ import py
+ import sys
+ def test_hello():
+ sys._gw = py.execnet.PopenGateway()
+ """, "-s", "--debug")
+ assert not p._gateways
+ assert py.std.sys._gw
+ py.test.raises(KeyError, "py.std.sys._gw.exit()") # already closed
+
Modified: py/trunk/py/test/plugin/pytest_plugintester.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_plugintester.py (original)
+++ py/trunk/py/test/plugin/pytest_plugintester.py Wed Mar 18 16:51:55 2009
@@ -155,8 +155,14 @@
def pyevent(self, eventname, *args, **kwargs):
""" called for each testing event. """
+ def pyevent_gateway_init(self, gateway):
+ """ called a gateway has been initialized. """
+
+ def pyevent_gateway_exit(self, gateway):
+ """ called when gateway is being exited. """
+
def pyevent_trace(self, category, msg):
- """ called for tracing events events. """
+ """ called for tracing events. """
def pyevent_internalerror(self, event):
""" called for internal errors. """
Modified: py/trunk/py/test/plugin/pytest_pytester.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_pytester.py (original)
+++ py/trunk/py/test/plugin/pytest_pytester.py Wed Mar 18 16:51:55 2009
@@ -21,6 +21,11 @@
def pytest_pyfuncarg_EventRecorder(self, pyfuncitem):
return EventRecorder
+ def pytest_pyfuncarg_eventrecorder(self, pyfuncitem):
+ evrec = EventRecorder(py._com.pyplugins)
+ pyfuncitem.addfinalizer(lambda: evrec.pyplugins.unregister(evrec))
+ return evrec
+
def test_generic(plugintester):
plugintester.apicheck(PytesterPlugin)
@@ -245,6 +250,8 @@
self.name = name
self.args = args
self.kwargs = kwargs
+ def __repr__(self):
+ return "<Event %r %r>" %(self.name, self.args)
class EventRecorder(object):
def __init__(self, pyplugins, debug=False): # True):
@@ -260,6 +267,13 @@
print "[event: %s]: %s **%s" %(name, ", ".join(map(str, args)), kwargs,)
self.events.append(Event(name, args, kwargs))
+ def popevent(self, name):
+ for i, event in py.builtin.enumerate(self.events):
+ if event.name == name:
+ del self.events[i]
+ return event
+ raise KeyError("popevent: %r not found in %r" %(name, self.events))
+
def get(self, cls):
l = []
for event in self.events:
Modified: py/trunk/py/test/pytestplugin.py
==============================================================================
--- py/trunk/py/test/pytestplugin.py (original)
+++ py/trunk/py/test/pytestplugin.py Wed Mar 18 16:51:55 2009
@@ -95,6 +95,7 @@
config = self._config
del self._config
self.pyplugins.call_each("pytest_unconfigure", config=config)
+ config.bus.unregister(self)
#
# XXX old code to automatically load classes
More information about the pytest-commit
mailing list