[py-svn] r63883 - in py/trunk/py: . test test/dist test/dist/testing test/looponfail test/plugin test/testing
hpk at codespeak.net
hpk at codespeak.net
Thu Apr 9 01:33:51 CEST 2009
Author: hpk
Date: Thu Apr 9 01:33:48 2009
New Revision: 63883
Modified:
py/trunk/py/_com.py
py/trunk/py/test/dist/dsession.py
py/trunk/py/test/dist/testing/test_dsession.py
py/trunk/py/test/dist/testing/test_nodemanage.py
py/trunk/py/test/dist/testing/test_txnode.py
py/trunk/py/test/dist/txnode.py
py/trunk/py/test/looponfail/remote.py
py/trunk/py/test/plugin/api.py
py/trunk/py/test/plugin/pytest__pytest.py
py/trunk/py/test/plugin/pytest_default.py
py/trunk/py/test/plugin/pytest_doctest.py
py/trunk/py/test/plugin/pytest_eventlog.py
py/trunk/py/test/plugin/pytest_pytester.py
py/trunk/py/test/plugin/pytest_resultlog.py
py/trunk/py/test/plugin/pytest_runner.py
py/trunk/py/test/plugin/pytest_terminal.py
py/trunk/py/test/pytestplugin.py
py/trunk/py/test/runner.py
py/trunk/py/test/session.py
py/trunk/py/test/testing/test_collect.py
py/trunk/py/test/testing/test_pytestplugin.py
py/trunk/py/test/testing/test_runner.py
py/trunk/py/test/testing/test_session.py
Log:
* moving many more events to become ordinary plugin hook calls.
* bit hackish because the code for handling the old events is
also still there ...
Modified: py/trunk/py/_com.py
==============================================================================
--- py/trunk/py/_com.py (original)
+++ py/trunk/py/_com.py Thu Apr 9 01:33:48 2009
@@ -115,11 +115,11 @@
def register(self, plugin):
assert not isinstance(plugin, str)
+ self.call_each("pytest_plugin_registered", plugin)
self._plugins.append(plugin)
- self.notify("plugin_registered", plugin)
def unregister(self, plugin):
- self.notify("plugin_unregistered", plugin)
+ self.call_each("pytest_plugin_unregistered", plugin)
self._plugins.remove(plugin)
def getplugins(self):
Modified: py/trunk/py/test/dist/dsession.py
==============================================================================
--- py/trunk/py/test/dist/dsession.py (original)
+++ py/trunk/py/test/dist/dsession.py Thu Apr 9 01:33:48 2009
@@ -24,20 +24,20 @@
self.shuttingdown = False
self.testsfailed = False
- def pyevent__itemtestreport(self, rep):
+ def pytest_itemtestreport(self, rep):
if rep.colitem in self.dsession.item2nodes:
self.dsession.removeitem(rep.colitem, rep.node)
if rep.failed:
self.testsfailed = True
- def pyevent__collectreport(self, rep):
+ def pytest_collectreport(self, rep):
if rep.passed:
self.colitems.extend(rep.result)
- def pyevent__testnodeready(self, node):
+ def pytest_testnodeready(self, node):
self.dsession.addnode(node)
- def pyevent__testnodedown(self, node, error=None):
+ def pytest_testnodedown(self, node, error=None):
pending = self.dsession.removenode(node)
if pending:
crashitem = pending[0]
@@ -95,8 +95,12 @@
continue
loopstate.dowork = True
- eventname, args, kwargs = eventcall
- self.bus.notify(eventname, *args, **kwargs)
+ callname, args, kwargs = eventcall
+ call = getattr(self.config.api, callname, None)
+ if call is not None:
+ call(*args, **kwargs)
+ else:
+ self.bus.notify(callname, *args, **kwargs)
# termination conditions
if ((loopstate.testsfailed and self.config.option.exitfirst) or
@@ -110,10 +114,9 @@
# once we are in shutdown mode we dont send
# events other than HostDown upstream
eventname, args, kwargs = self.queue.get()
- if eventname == "testnodedown":
- node, error = args[0], args[1]
- self.bus.notify("testnodedown", node, error)
- self.removenode(node)
+ if eventname == "pytest_testnodedown":
+ self.config.api.pytest_testnodedown(**kwargs)
+ self.removenode(kwargs['node'])
if not self.node2pending:
# finished
if loopstate.testsfailed:
@@ -174,8 +177,8 @@
if isinstance(next, py.test.collect.Item):
senditems.append(next)
else:
- self.bus.notify("collectionstart", next)
- self.queueevent("collectreport", basic_collect_report(next))
+ self.bus.notify("collectstart", next)
+ self.queueevent("pytest_collectreport", basic_collect_report(next))
if self.config.option.dist == "each":
self.senditems_each(senditems)
else:
@@ -197,7 +200,7 @@
pending.extend(sending)
for item in sending:
self.item2nodes.setdefault(item, []).append(node)
- self.bus.notify("itemstart", item, node)
+ self.config.api.pytest_itemstart(item=item, node=node)
tosend[:] = tosend[room:] # update inplace
if tosend:
# we have some left, give it to the main loop
@@ -216,7 +219,7 @@
# "sending same item %r to multiple "
# "not implemented" %(item,))
self.item2nodes.setdefault(item, []).append(node)
- self.bus.notify("itemstart", item, node)
+ self.config.api.pytest_itemstart(item=item, node=node)
pending.extend(sending)
tosend[:] = tosend[room:] # update inplace
if not tosend:
@@ -239,7 +242,7 @@
longrepr = "!!! Node %r crashed during running of test %r" %(node, item)
rep = ItemTestReport(item, when="???", excinfo=longrepr)
rep.node = node
- self.bus.notify("itemtestreport", rep)
+ self.config.api.pytest_itemtestreport(rep=rep)
def setup(self):
""" setup any neccessary resources ahead of the test run. """
Modified: py/trunk/py/test/dist/testing/test_dsession.py
==============================================================================
--- py/trunk/py/test/dist/testing/test_dsession.py (original)
+++ py/trunk/py/test/dist/testing/test_dsession.py Thu Apr 9 01:33:48 2009
@@ -80,7 +80,7 @@
session = DSession(modcol.config)
session.triggertesting([modcol])
name, args, kwargs = session.queue.get(block=False)
- assert name == 'collectreport'
+ assert name == 'pytest_collectreport'
rep, = args
assert len(rep.result) == 1
@@ -135,7 +135,7 @@
session.queueevent("anonymous")
session.loop_once(loopstate)
assert node.sent == [[item]]
- session.queueevent("itemtestreport", run(item, node))
+ session.queueevent("pytest_itemtestreport", run(item, node))
session.loop_once(loopstate)
assert loopstate.shuttingdown
assert not loopstate.testsfailed
@@ -148,7 +148,7 @@
session.addnode(node)
# setup a HostDown event
- session.queueevent("testnodedown", node, None)
+ session.queueevent("pytest_testnodedown", node, None)
loopstate = session._initloopstate([item])
loopstate.dowork = False
@@ -156,7 +156,7 @@
dumpqueue(session.queue)
assert loopstate.exitstatus == outcome.EXIT_NOHOSTS
- def test_testnodedown_causes_reschedule_pending(self, testdir, EventRecorder):
+ def test_testnodedown_causes_reschedule_pending(self, testdir):
modcol = testdir.getmodulecol("""
def test_crash():
assert 0
@@ -174,8 +174,8 @@
# have one test pending for a node that goes down
session.senditems_load([item1, item2])
node = session.item2nodes[item1] [0]
- session.queueevent("testnodedown", node, None)
- evrec = EventRecorder(session.bus)
+ session.queueevent("pytest_testnodedown", node, None)
+ evrec = testdir.geteventrecorder(session.bus)
print session.item2nodes
loopstate = session._initloopstate([])
session.loop_once(loopstate)
@@ -192,18 +192,18 @@
# setup a session with two nodes
session = DSession(item.config)
node1 = MockNode()
- session.queueevent("testnodeready", node1)
+ session.queueevent("pytest_testnodeready", node1)
loopstate = session._initloopstate([item])
loopstate.dowork = False
assert len(session.node2pending) == 0
session.loop_once(loopstate)
assert len(session.node2pending) == 1
- def test_event_propagation(self, testdir, EventRecorder):
+ def test_event_propagation(self, testdir):
item = testdir.getitem("def test_func(): pass")
session = DSession(item.config)
- evrec = EventRecorder(session.bus)
+ evrec = testdir.geteventrecorder(session.bus)
session.queueevent("NOP", 42)
session.loop_once(session._initloopstate([]))
assert evrec.getcall('NOP')
@@ -219,10 +219,10 @@
assert node.sent == [[item]]
ev = run(item, node)
- session.queueevent("itemtestreport", ev)
+ session.queueevent("pytest_itemtestreport", rep=ev)
session.loop_once(loopstate)
assert loopstate.shuttingdown
- session.queueevent("testnodedown", node, None)
+ session.queueevent("pytest_testnodedown", node=node, error=None)
session.loop_once(loopstate)
dumpqueue(session.queue)
return session, loopstate.exitstatus
@@ -256,30 +256,30 @@
# run tests ourselves and produce reports
ev1 = run(items[0], node)
ev2 = run(items[1], node)
- session.queueevent("itemtestreport", ev1) # a failing one
- session.queueevent("itemtestreport", ev2)
+ session.queueevent("pytest_itemtestreport", rep=ev1) # a failing one
+ session.queueevent("pytest_itemtestreport", rep=ev2)
# now call the loop
loopstate = session._initloopstate(items)
session.loop_once(loopstate)
assert loopstate.testsfailed
assert loopstate.shuttingdown
- def test_shuttingdown_filters_events(self, testdir, EventRecorder):
+ def test_shuttingdown_filters_events(self, testdir):
item = testdir.getitem("def test_func(): pass")
session = DSession(item.config)
node = MockNode()
session.addnode(node)
loopstate = session._initloopstate([])
loopstate.shuttingdown = True
- evrec = EventRecorder(session.bus)
- session.queueevent("itemtestreport", run(item, node))
+ evrec = testdir.geteventrecorder(session.bus)
+ session.queueevent("pytest_itemtestreport", rep=run(item, node))
session.loop_once(loopstate)
- assert not evrec.getcalls("testnodedown")
- session.queueevent("testnodedown", node, None)
+ assert not evrec.getcalls("pytest_testnodedown")
+ session.queueevent("pytest_testnodedown", node=node, error=None)
session.loop_once(loopstate)
- assert evrec.getcall('testnodedown').node == node
+ assert evrec.getcall('pytest_testnodedown').node == node
- def test_filteritems(self, testdir, EventRecorder):
+ def test_filteritems(self, testdir):
modcol = testdir.getmodulecol("""
def test_fail():
assert 0
@@ -292,7 +292,7 @@
dsel = session.filteritems([modcol])
assert dsel == [modcol]
items = modcol.collect()
- evrec = EventRecorder(session.bus)
+ evrec = testdir.geteventrecorder(session.bus)
remaining = session.filteritems(items)
assert remaining == []
@@ -313,13 +313,13 @@
node = MockNode()
session.addnode(node)
session.senditems_load([item])
- session.queueevent("itemtestreport", run(item, node))
+ session.queueevent("pytest_itemtestreport", rep=run(item, node))
loopstate = session._initloopstate([])
session.loop_once(loopstate)
assert node._shutdown is True
assert loopstate.exitstatus is None, "loop did not wait for testnodedown"
assert loopstate.shuttingdown
- session.queueevent("testnodedown", node, None)
+ session.queueevent("pytest_testnodedown", node=node, error=None)
session.loop_once(loopstate)
assert loopstate.exitstatus == 0
@@ -340,10 +340,10 @@
# node2pending will become empty when the loop sees the report
rep = run(item1, node)
- session.queueevent("itemtestreport", run(item1, node))
+ session.queueevent("pytest_itemtestreport", rep=run(item1, node))
# but we have a collection pending
- session.queueevent("collectreport", colreport)
+ session.queueevent("pytest_collectreport", rep=colreport)
loopstate = session._initloopstate([])
session.loop_once(loopstate)
@@ -354,7 +354,6 @@
assert loopstate.exitstatus is None, "loop did not care for colitems"
def test_dist_some_tests(self, testdir):
- from py.__.test.dist.testing.test_txnode import EventQueue
p1 = testdir.makepyfile(test_one="""
def test_1():
pass
@@ -366,16 +365,16 @@
""")
config = testdir.parseconfig('-d', p1, '--tx=popen')
dsession = DSession(config)
- eq = EventQueue(config.bus)
+ callrecorder = testdir.geteventrecorder(config.bus).callrecorder
dsession.main([config.getfsnode(p1)])
- ev, = eq.geteventargs("itemtestreport")
- assert ev.passed
- ev, = eq.geteventargs("itemtestreport")
- assert ev.skipped
- ev, = eq.geteventargs("itemtestreport")
- assert ev.failed
+ rep = callrecorder.popcall("pytest_itemtestreport").rep
+ assert rep.passed
+ rep = callrecorder.popcall("pytest_itemtestreport").rep
+ assert rep.skipped
+ rep = callrecorder.popcall("pytest_itemtestreport").rep
+ assert rep.failed
# see that the node is really down
- node, error = eq.geteventargs("testnodedown")
+ node = callrecorder.popcall("pytest_testnodedown").node
assert node.gateway.spec.popen
- eq.geteventargs("testrunfinish")
+ #XXX eq.geteventargs("pytest_testrunfinish")
Modified: py/trunk/py/test/dist/testing/test_nodemanage.py
==============================================================================
--- py/trunk/py/test/dist/testing/test_nodemanage.py (original)
+++ py/trunk/py/test/dist/testing/test_nodemanage.py Thu Apr 9 01:33:48 2009
@@ -97,14 +97,14 @@
assert gwspec._samefilesystem()
assert not gwspec.chdir
- def test_setup_DEBUG(self, source, EventRecorder):
+ def test_setup_DEBUG(self, source, testdir):
specs = ["popen"] * 2
source.join("conftest.py").write("rsyncdirs = ['a']")
source.ensure('a', dir=1)
config = py.test.config._reparse([source, '--debug'])
assert config.option.debug
nodemanager = NodeManager(config, specs)
- sorter = EventRecorder(config.bus, debug=True)
+ sorter = testdir.geteventrecorder(config.bus)
nodemanager.setup_nodes(putevent=[].append)
for spec in nodemanager.gwmanager.specs:
l = sorter.getcalls("trace")
@@ -119,6 +119,6 @@
""")
sorter = testdir.inline_run("-d", "--rsyncdir=%s" % testdir.tmpdir,
"--tx=%s" % specssh, testdir.tmpdir)
- ev = sorter.getfirstnamed("itemtestreport")
+ ev = sorter.getfirstnamed(pytest_itemtestreport)
assert ev.passed
Modified: py/trunk/py/test/dist/testing/test_txnode.py
==============================================================================
--- py/trunk/py/test/dist/testing/test_txnode.py (original)
+++ py/trunk/py/test/dist/testing/test_txnode.py Thu Apr 9 01:33:48 2009
@@ -26,7 +26,9 @@
name, args, kwargs = eventcall
assert isinstance(name, str)
if name == eventname:
- return args
+ if args:
+ return args
+ return kwargs
events.append(name)
if name == "internalerror":
print str(kwargs["excrepr"])
@@ -78,9 +80,9 @@
def test_crash_invalid_item(self, mysetup):
node = mysetup.makenode()
node.send(123) # invalid item
- n, error = mysetup.geteventargs("testnodedown")
- assert n is node
- assert str(error).find("AttributeError") != -1
+ kwargs = mysetup.geteventargs("pytest_testnodedown")
+ assert kwargs['node'] is node
+ assert str(kwargs['error']).find("AttributeError") != -1
def test_crash_killed(self, testdir, mysetup):
if not hasattr(py.std.os, 'kill'):
@@ -92,16 +94,16 @@
""")
node = mysetup.makenode(item.config)
node.send(item)
- n, error = mysetup.geteventargs("testnodedown")
- assert n is node
- assert str(error).find("Not properly terminated") != -1
+ kwargs = mysetup.geteventargs("pytest_testnodedown")
+ assert kwargs['node'] is node
+ assert str(kwargs['error']).find("Not properly terminated") != -1
def test_node_down(self, mysetup):
node = mysetup.makenode()
node.shutdown()
- n, error = mysetup.geteventargs("testnodedown")
- assert n is node
- assert not error
+ kwargs = mysetup.geteventargs("pytest_testnodedown")
+ assert kwargs['node'] is node
+ assert not kwargs['error']
node.callback(node.ENDMARK)
excinfo = py.test.raises(IOError,
"mysetup.geteventargs('testnodedown', timeout=0.01)")
@@ -118,11 +120,11 @@
item = testdir.getitem("def test_func(): pass")
node = mysetup.makenode(item.config)
node.send(item)
- ev, = mysetup.geteventargs("itemtestreport")
- assert ev.passed
- assert ev.colitem == item
- #assert event.item == item
- #assert event.item is not item
+ kwargs = mysetup.geteventargs("pytest_itemtestreport")
+ rep = kwargs['rep']
+ assert rep.passed
+ print rep
+ assert rep.colitem == item
def test_send_some(self, testdir, mysetup):
items = testdir.getitems("""
@@ -138,10 +140,11 @@
for item in items:
node.send(item)
for outcome in "passed failed skipped".split():
- ev, = mysetup.geteventargs("itemtestreport")
- assert getattr(ev, outcome)
+ kwargs = mysetup.geteventargs("pytest_itemtestreport")
+ rep = kwargs['rep']
+ assert getattr(rep, outcome)
node.sendlist(items)
for outcome in "passed failed skipped".split():
- ev, = mysetup.geteventargs("itemtestreport")
- assert getattr(ev, outcome)
+ rep = mysetup.geteventargs("pytest_itemtestreport")['rep']
+ assert getattr(rep, outcome)
Modified: py/trunk/py/test/dist/txnode.py
==============================================================================
--- py/trunk/py/test/dist/txnode.py (original)
+++ py/trunk/py/test/dist/txnode.py Thu Apr 9 01:33:48 2009
@@ -38,21 +38,21 @@
if not self._down:
if not err:
err = "Not properly terminated"
- self.notify("testnodedown", self, err)
+ self.notify("pytest_testnodedown", node=self, error=err)
self._down = True
return
eventname, args, kwargs = eventcall
if eventname == "slaveready":
if self._sendslaveready:
self._sendslaveready(self)
- self.notify("testnodeready", self)
+ self.notify("pytest_testnodeready", node=self)
elif eventname == "slavefinished":
self._down = True
- self.notify("testnodedown", self, None)
- elif eventname == "itemtestreport":
- rep = args[0]
+ self.notify("pytest_testnodedown", error=None, node=self)
+ elif eventname == "pytest_itemtestreport":
+ rep = kwargs['rep']
rep.node = self
- self.notify("itemtestreport", rep)
+ self.notify("pytest_itemtestreport", rep=rep)
else:
self.notify(eventname, *args, **kwargs)
except KeyboardInterrupt:
@@ -104,8 +104,8 @@
def sendevent(self, eventname, *args, **kwargs):
self.channel.send((eventname, args, kwargs))
- def pyevent__itemtestreport(self, rep):
- self.sendevent("itemtestreport", rep)
+ def pytest_itemtestreport(self, rep):
+ self.sendevent("pytest_itemtestreport", rep=rep)
def run(self):
channel = self.channel
Modified: py/trunk/py/test/looponfail/remote.py
==============================================================================
--- py/trunk/py/test/looponfail/remote.py (original)
+++ py/trunk/py/test/looponfail/remote.py Thu Apr 9 01:33:48 2009
@@ -137,10 +137,10 @@
session.shouldclose = channel.isclosed
class Failures(list):
- def pyevent__itemtestreport(self, rep):
+ def pytest_itemtestreport(self, rep):
if rep.failed:
self.append(rep)
- pyevent__collectreport = pyevent__itemtestreport
+ pytest_collectreport = pytest_itemtestreport
failreports = Failures()
session.bus.register(failreports)
Modified: py/trunk/py/test/plugin/api.py
==============================================================================
--- py/trunk/py/test/plugin/api.py (original)
+++ py/trunk/py/test/plugin/api.py Thu Apr 9 01:33:48 2009
@@ -68,6 +68,36 @@
""" return processed content for a given doctest"""
pytest_doctest_prepare_content.firstresult = True
+ def pytest_itemstart(self, item, node=None):
+ """ test item gets collected. """
+
+ def pytest_itemtestreport(self, rep):
+ """ test has been run. """
+
+ def pytest_item_runtest_finished(self, item, excinfo, outerr):
+ """ test has been run. """
+
+ def pytest_itemsetupreport(self, rep):
+ """ a report on running a fixture function. """
+
+ def pytest_collectstart(self, collector):
+ """ collector starts collecting. """
+
+ def pytest_collectreport(self, rep):
+ """ collector finished collecting. """
+
+ def pytest_plugin_registered(self, plugin):
+ """ a new py lib plugin got registered. """
+
+ def pytest_plugin_unregistered(self, plugin):
+ """ a py lib plugin got unregistered. """
+
+ def pytest_testnodeready(self, node):
+ """ Test Node is ready to operate. """
+
+ def pytest_testnodedown(self, node, error):
+ """ Test Node is down. """
+
class Events:
# Events
@@ -83,26 +113,9 @@
def pyevent__internalerror(self, excrepr):
""" called for internal errors. """
- def pyevent__itemstart(self, item, node=None):
- """ test item gets collected. """
-
- def pyevent__itemtestreport(self, rep):
- """ test has been run. """
-
- def pyevent__item_runtest_finished(self, item, excinfo, outerr):
- """ test has been run. """
-
- def pyevent__itemsetupreport(self, rep):
- """ a report on running a fixture function. """
-
def pyevent__deselected(self, items):
""" collected items that were deselected (by keyword). """
- def pyevent__collectionstart(self, collector):
- """ collector starts collecting. """
-
- def pyevent__collectreport(self, rep):
- """ collector finished collecting. """
def pyevent__testrunstart(self):
""" whole test run starts. """
@@ -115,19 +128,11 @@
The gateway will have an 'id' attribute that is unique
within the gateway manager context.
"""
- def pyevent__testnodeready(self, node):
- """ Test Node is ready to operate. """
-
- def pyevent__testnodedown(self, node, error):
- """ Test Node is down. """
- def pyevent__rescheduleitems(self, items):
+ def pytest_rescheduleitems(self, items):
""" reschedule Items from a node that went down. """
def pyevent__looponfailinfo(self, failreports, rootdirs):
""" info for repeating failing tests. """
- def pyevent__plugin_registered(self, plugin):
- """ a new py lib plugin got registered. """
-
Modified: py/trunk/py/test/plugin/pytest__pytest.py
==============================================================================
--- py/trunk/py/test/plugin/pytest__pytest.py (original)
+++ py/trunk/py/test/plugin/pytest__pytest.py Thu Apr 9 01:33:48 2009
@@ -21,10 +21,13 @@
def __init__(self, name, locals):
assert '_name' not in locals
self.__dict__.update(locals)
+ self.__dict__.pop('self')
self._name = name
def __repr__(self):
- return "<ParsedCall %r>" %(self.__dict__,)
+ d = self.__dict__.copy()
+ del d['_name']
+ return "<ParsedCall %r(**%r)>" %(self._name, d)
class CallRecorder:
def __init__(self, pyplugins):
@@ -47,6 +50,11 @@
for recorder in self._recorders.values():
self._pyplugins.unregister(recorder)
+ def recordsmethod(self, name):
+ for apiclass in self._recorders:
+ if hasattr(apiclass, name):
+ return True
+
def _getcallparser(self, method):
name = method.__name__
args, varargs, varkw, default = py.std.inspect.getargspec(method)
@@ -70,6 +78,22 @@
return call
raise ValueError("could not find call %r in %r" %(name, self.calls))
+ def getcalls(self, names):
+ if isinstance(names, str):
+ names = names.split()
+ for name in names:
+ for cls in self._recorders:
+ if name in vars(cls):
+ break
+ else:
+ raise ValueError("callname %r not found in %r" %(
+ name, self._recorders.keys()))
+ l = []
+ for call in self.calls:
+ if call._name in names:
+ l.append(call)
+ return l
+
def test_generic(plugintester):
plugintester.apicheck(_pytestPlugin)
Modified: py/trunk/py/test/plugin/pytest_default.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_default.py (original)
+++ py/trunk/py/test/plugin/pytest_default.py Thu Apr 9 01:33:48 2009
@@ -10,7 +10,7 @@
else:
runner = basic_run_report
report = runner(item, pdb=pdb)
- item.config.pytestplugins.notify("itemtestreport", report)
+ item.config.api.pytest_itemtestreport(rep=report)
return True
def pytest_item_makereport(self, item, excinfo, when, outerr):
@@ -20,8 +20,9 @@
def pytest_item_runtest_finished(self, item, excinfo, outerr):
from py.__.test import runner
rep = runner.ItemTestReport(item, excinfo, "execute", outerr)
- item.config.pytestplugins.notify("itemtestreport", rep)
+ item.config.api.pytest_itemtestreport(rep=rep)
+ # XXX make this access pyfuncitem.args or funcargs
def pytest_pyfunc_call(self, pyfuncitem, args, kwargs):
pyfuncitem.obj(*args, **kwargs)
Modified: py/trunk/py/test/plugin/pytest_doctest.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_doctest.py (original)
+++ py/trunk/py/test/plugin/pytest_doctest.py Thu Apr 9 01:33:48 2009
@@ -122,7 +122,7 @@
2
""")
sorter = testdir.inline_run(p)
- call = sorter.getcall("itemtestreport")
+ call = sorter.getcall("pytest_itemtestreport")
assert call.rep.failed
assert call.rep.longrepr
# XXX
Modified: py/trunk/py/test/plugin/pytest_eventlog.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_eventlog.py (original)
+++ py/trunk/py/test/plugin/pytest_eventlog.py Thu Apr 9 01:33:48 2009
@@ -26,6 +26,7 @@
# plugin tests
# ===============================================================================
+ at py.test.mark.xfail
def test_generic(plugintester):
plugintester.apicheck(EventlogPlugin)
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 Thu Apr 9 01:33:48 2009
@@ -6,8 +6,10 @@
import inspect
from py.__.test import runner
from py.__.test.config import Config as pytestConfig
+from pytest__pytest import CallRecorder
import api
+
class PytesterPlugin:
def pytest_funcarg__linecomp(self, pyfuncitem):
return LineComp()
@@ -19,8 +21,8 @@
tmptestdir = TmpTestdir(pyfuncitem)
return tmptestdir
- def pytest_funcarg__EventRecorder(self, pyfuncitem):
- return EventRecorder
+ #def pytest_funcarg__EventRecorder(self, pyfuncitem):
+ # return EventRecorder
def pytest_funcarg__eventrecorder(self, pyfuncitem):
evrec = EventRecorder(py._com.pyplugins)
@@ -74,10 +76,12 @@
if hasattr(self, '_olddir'):
self._olddir.chdir()
- def geteventrecorder(self, config):
- evrec = EventRecorder(config.bus)
- self.pyfuncitem.addfinalizer(lambda: config.bus.unregister(evrec))
- return evrec
+ def geteventrecorder(self, bus):
+ sorter = EventRecorder(bus)
+ sorter.callrecorder = CallRecorder(bus)
+ sorter.callrecorder.start_recording(api.PluginHooks)
+ self.pyfuncitem.addfinalizer(sorter.callrecorder.finalize)
+ return sorter
def chdir(self):
old = self.tmpdir.chdir()
@@ -128,7 +132,7 @@
#config = self.parseconfig(*args)
config = self.parseconfig(*args)
session = config.initsession()
- rec = EventRecorder(config.bus)
+ rec = self.geteventrecorder(config.bus)
colitems = [config.getfsnode(arg) for arg in config.args]
items = list(session.genitems(colitems))
return items, rec
@@ -150,10 +154,10 @@
config = self.parseconfig(*args)
config.pytestplugins.do_configure(config)
session = config.initsession()
- sorter = EventRecorder(config.bus)
+ sorter = self.geteventrecorder(config.bus)
session.main()
config.pytestplugins.do_unconfigure(config)
- return sorter
+ return sorter
def config_preparse(self):
config = self.Config()
@@ -306,11 +310,17 @@
if len(names) == 1 and isinstance(names, str):
names = names.split()
l = []
- for event in self.events:
- if event.name in names:
- method = self._getcallparser(event.name)
- pevent = method(*event.args, **event.kwargs)
- l.append(pevent)
+ for name in names:
+ if self.callrecorder.recordsmethod("pytest_" + name):
+ name = "pytest_" + name
+ if self.callrecorder.recordsmethod(name):
+ l.extend(self.callrecorder.getcalls(name))
+ else:
+ for event in self.events:
+ if event.name == name:
+ method = self._getcallparser(event.name)
+ pevent = method(*event.args, **event.kwargs)
+ l.append(pevent)
return l
def _getcallparser(self, eventname):
@@ -328,7 +338,7 @@
# functionality for test reports
def getreports(self, names="itemtestreport collectreport"):
- names = names.split()
+ names = [("pytest_" + x) for x in names.split()]
l = []
for call in self.getcalls(*names):
l.append(call.rep)
@@ -386,6 +396,7 @@
def unregister(self):
self.pyplugins.unregister(self)
+ at py.test.mark.xfail
def test_eventrecorder():
bus = py._com.PyPlugins()
recorder = EventRecorder(bus)
@@ -395,7 +406,7 @@
rep = runner.ItemTestReport(None, None)
rep.passed = False
rep.failed = True
- bus.notify("itemtestreport", rep)
+ bus.notify("pytest_itemtestreport", rep)
failures = recorder.getfailures()
assert failures == [rep]
failures = recorder.getfailures()
@@ -404,12 +415,12 @@
rep = runner.ItemTestReport(None, None)
rep.passed = False
rep.skipped = True
- bus.notify("itemtestreport", rep)
+ bus.notify("pytest_itemtestreport", rep)
rep = runner.CollectReport(None, None)
rep.passed = False
rep.failed = True
- bus.notify("itemtestreport", rep)
+ bus.notify("pytest_itemtestreport", rep)
passed, skipped, failed = recorder.listoutcomes()
assert not passed and skipped and failed
@@ -423,7 +434,7 @@
recorder.clear()
assert not recorder.events
assert not recorder.getfailures()
- bus.notify("itemtestreport", rep)
+ bus.notify(pytest_itemtestreport, rep)
assert not recorder.events
assert not recorder.getfailures()
Modified: py/trunk/py/test/plugin/pytest_resultlog.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_resultlog.py (original)
+++ py/trunk/py/test/plugin/pytest_resultlog.py Thu Apr 9 01:33:48 2009
@@ -58,7 +58,7 @@
testpath = generic_path(event.colitem)
self.write_log_entry(testpath, shortrepr, longrepr)
- def pyevent__itemtestreport(self, rep):
+ def pytest_itemtestreport(self, rep):
code = rep.shortrepr
if rep.passed:
longrepr = ""
@@ -68,7 +68,7 @@
longrepr = str(rep.longrepr.reprcrash.message)
self.log_outcome(rep, code, longrepr)
- def pyevent__collectreport(self, rep):
+ def pytest_collectreport(self, rep):
if not rep.passed:
if rep.failed:
code = "F"
Modified: py/trunk/py/test/plugin/pytest_runner.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_runner.py (original)
+++ py/trunk/py/test/plugin/pytest_runner.py Thu Apr 9 01:33:48 2009
@@ -13,7 +13,7 @@
call = item.config.guardedcall(lambda: setupstate.prepare(item))
rep = ItemSetupReport(item, call.excinfo, call.outerr)
if call.excinfo:
- item.config.pytestplugins.notify("itemsetupreport", rep)
+ item.config.pytestplugins.notify(pytest_itemsetupreport, rep)
else:
call = item.config.guardedcall(lambda: item.runtest())
item.config.api.pytest_item_runtest_finished(
@@ -21,7 +21,7 @@
call = item.config.guardedcall(lambda: self.teardown_exact(item))
if call.excinfo:
rep = ItemSetupReport(item, call.excinfo, call.outerr)
- item.config.pytestplugins.notify("itemsetupreport", rep)
+ item.config.api.pytest_itemsetupreport(rep=rep)
def pytest_collector_collect(self, collector):
call = item.config.guardedcall(lambda x: collector._memocollect())
@@ -208,9 +208,9 @@
item = testdir.getitem("""def test_func(): pass""")
plugin = RunnerPlugin()
plugin.pytest_configure(item.config)
- sorter = testdir.geteventrecorder(item.config)
+ sorter = testdir.geteventrecorder(item.config.bus)
plugin.pytest_item_setup_and_runtest(item)
- rep = sorter.getcall("itemtestreport").rep
+ rep = sorter.getcall("pytest_itemtestreport").rep
assert rep.passed
class TestSetupEvents:
@@ -223,7 +223,7 @@
def test_func():
pass
""")
- assert not sorter.getcalls("itemsetupreport")
+ assert not sorter.getcalls(pytest_itemsetupreport)
def test_setup_fails(self, testdir):
sorter = testdir.inline_runsource("""
@@ -233,7 +233,7 @@
def test_func():
pass
""")
- rep = sorter.popcall("itemsetupreport").rep
+ rep = sorter.popcall(pytest_itemsetupreport).rep
assert rep.failed
assert not rep.skipped
assert rep.excrepr
@@ -248,7 +248,7 @@
print "13"
raise ValueError(25)
""")
- rep = evrec.popcall("itemsetupreport").rep
+ rep = evrec.popcall(pytest_itemsetupreport).rep
assert rep.failed
assert rep.item == item
assert not rep.passed
@@ -263,7 +263,7 @@
def test_func():
pass
""")
- rep = sorter.popcall("itemsetupreport")
+ rep = sorter.popcall(pytest_itemsetupreport)
assert not rep.failed
assert rep.skipped
assert rep.excrepr
Modified: py/trunk/py/test/plugin/pytest_terminal.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_terminal.py (original)
+++ py/trunk/py/test/plugin/pytest_terminal.py Thu Apr 9 01:33:48 2009
@@ -116,7 +116,7 @@
targets = ", ".join([gw.id for gw in gateways])
self.write_line("rsyncfinish: %s -> %s" %(source, targets))
- def pyevent__plugin_registered(self, plugin):
+ def pytest_plugin_registered(self, plugin):
if self.config.option.traceconfig:
msg = "PLUGIN registered: %s" %(plugin,)
# XXX this event may happen during setup/teardown time
@@ -124,10 +124,10 @@
# which garbles our output if we use self.write_line
self.write_line(msg)
- def pyevent__testnodeready(self, node):
+ def pytest_testnodeready(self, node):
self.write_line("%s txnode ready to receive tests" %(node.gateway.id,))
- def pyevent__testnodedown(self, node, error):
+ def pytest_testnodedown(self, node, error):
if error:
self.write_line("%s node down, error: %s" %(node.gateway.id, error))
@@ -136,7 +136,7 @@
self.config.option.traceconfig and category.find("config") != -1:
self.write_line("[%s] %s" %(category, msg))
- def pyevent__itemstart(self, item, node=None):
+ def pytest_itemstart(self, item, node=None):
if self.config.option.debug:
info = item.repr_metainfo()
line = info.verboseline(basedir=self.curdir) + " "
@@ -154,14 +154,14 @@
#self.write_fspath_result(fspath, "")
self.write_ensure_prefix(line, "")
- def pyevent__rescheduleitems(self, items):
+ def pytest_rescheduleitems(self, items):
if self.config.option.debug:
self.write_sep("!", "RESCHEDULING %s " %(items,))
def pyevent__deselected(self, items):
self.stats.setdefault('deselected', []).append(items)
- def pyevent__itemtestreport(self, rep):
+ def pytest_itemtestreport(self, rep):
fspath = rep.colitem.fspath
cat, letter, word = self.getcategoryletterword(rep)
if isinstance(word, tuple):
@@ -184,7 +184,7 @@
self._tw.write(" " + line)
self.currentfspath = -2
- def pyevent__collectreport(self, rep):
+ def pytest_collectreport(self, rep):
if not rep.passed:
if rep.failed:
self.stats.setdefault("failed", []).append(rep)
@@ -309,14 +309,14 @@
def outindent(self, line):
self.out.line(self.indent + str(line))
- def pyevent__collectionstart(self, collector):
+ def pytest_collectstart(self, collector):
self.outindent(collector)
self.indent += self.INDENT
- def pyevent__itemstart(self, item, node=None):
+ def pytest_itemstart(self, item, node=None):
self.outindent(item)
- def pyevent__collectreport(self, rep):
+ def pytest_collectreport(self, rep):
if not rep.passed:
self.outindent("!!! %s !!!" % rep.longrepr.reprcrash.message)
self._failed.append(rep)
@@ -373,7 +373,7 @@
for item in testdir.genitems([modcol]):
ev = runner.basic_run_report(item)
- rep.config.bus.notify("itemtestreport", ev)
+ rep.config.api.pytest_itemtestreport(rep=ev)
linecomp.assert_contains_lines([
"*test_pass_skip_fail.py .sF"
])
@@ -400,10 +400,10 @@
items = modcol.collect()
rep.config.option.debug = True #
for item in items:
- rep.config.bus.notify("itemstart", item, None)
+ rep.config.api.pytest_itemstart(item=item, node=None)
s = linecomp.stringio.getvalue().strip()
assert s.endswith(item.name)
- rep.config.bus.notify("itemtestreport", runner.basic_run_report(item))
+ rep.config.api.pytest_itemtestreport(rep=runner.basic_run_report(item))
linecomp.assert_contains_lines([
"*test_pass_skip_fail_verbose.py:2: *test_ok*PASS*",
@@ -520,8 +520,8 @@
rep.config.bus.notify("testrunstart")
rep.config.bus.notify("testrunstart")
for item in testdir.genitems([modcol]):
- rep.config.bus.notify("itemtestreport",
- runner.basic_run_report(item))
+ rep.config.api.pytest_itemtestreport(
+ rep=runner.basic_run_report(item))
rep.config.bus.notify("testrunfinish", exitstatus=1)
s = linecomp.stringio.getvalue()
if tbopt == "long":
@@ -548,7 +548,7 @@
l = list(testdir.genitems([modcol]))
assert len(l) == 1
modcol.config.option.debug = True
- rep.config.bus.notify("itemstart", l[0])
+ rep.config.api.pytest_itemstart(item=l[0])
linecomp.assert_contains_lines([
"*test_show_path_before_running_test.py*"
])
@@ -569,8 +569,8 @@
bus.notify("testrunstart")
try:
for item in testdir.genitems([modcol]):
- bus.notify("itemtestreport",
- runner.basic_run_report(item))
+ modcol.config.api.pytest_itemtestreport(
+ rep=runner.basic_run_report(item))
except KeyboardInterrupt:
excinfo = py.code.ExceptionInfo()
else:
@@ -628,17 +628,17 @@
rep = CollectonlyReporter(modcol.config, out=linecomp.stringio)
modcol.config.bus.register(rep)
indent = rep.indent
- rep.config.bus.notify("collectionstart", modcol)
+ rep.config.api.pytest_collectstart(collector=modcol)
linecomp.assert_contains_lines([
"<Module 'test_collectonly_basic.py'>"
])
item = modcol.join("test_func")
- rep.config.bus.notify("itemstart", item)
+ rep.config.api.pytest_itemstart(item=item)
linecomp.assert_contains_lines([
" <Function 'test_func'>",
])
- rep.config.bus.notify( "collectreport",
- runner.CollectReport(modcol, [], excinfo=None))
+ rep.config.api.pytest_collectreport(
+ rep=runner.CollectReport(modcol, [], excinfo=None))
assert rep.indent == indent
def test_collectonly_skipped_module(self, testdir, linecomp):
Modified: py/trunk/py/test/pytestplugin.py
==============================================================================
--- py/trunk/py/test/pytestplugin.py (original)
+++ py/trunk/py/test/pytestplugin.py Thu Apr 9 01:33:48 2009
@@ -86,14 +86,14 @@
if excinfo is None:
excinfo = py.code.ExceptionInfo()
excrepr = excinfo.getrepr(funcargs=True, showlocals=True)
- return self.notify("internalerror", excrepr)
+ return self.notify("internalerror", excrepr=excrepr)
def do_addoption(self, parser):
methods = self.pyplugins.listattr("pytest_addoption", reverse=True)
mc = py._com.MultiCall(methods, parser=parser)
mc.execute()
- def pyevent__plugin_registered(self, plugin):
+ def pytest_plugin_registered(self, plugin):
if hasattr(self, '_config'):
self.pyplugins.call_plugin(plugin, "pytest_addoption", parser=self._config._parser)
self.pyplugins.call_plugin(plugin, "pytest_configure", config=self._config)
Modified: py/trunk/py/test/runner.py
==============================================================================
--- py/trunk/py/test/runner.py (original)
+++ py/trunk/py/test/runner.py Thu Apr 9 01:33:48 2009
@@ -100,6 +100,7 @@
class ItemTestReport(BaseReport):
failed = passed = skipped = False
+ # XXX rename colitem to item here
def __init__(self, colitem, excinfo=None, when=None, outerr=None):
self.colitem = colitem
if colitem and when != "setup":
Modified: py/trunk/py/test/session.py
==============================================================================
--- py/trunk/py/test/session.py (original)
+++ py/trunk/py/test/session.py Thu Apr 9 01:33:48 2009
@@ -34,20 +34,19 @@
colitems[:] = list(next) + colitems
continue
assert self.bus is next.config.bus
- notify = self.bus.notify
if isinstance(next, Item):
remaining = self.filteritems([next])
if remaining:
- notify("itemstart", next)
+ self.config.api.pytest_itemstart(item=next)
yield next
else:
assert isinstance(next, Collector)
- notify("collectionstart", next)
+ self.config.api.pytest_collectstart(collector=next)
rep = basic_collect_report(next)
if rep.passed:
for x in self.genitems(rep.result, keywordexpr):
yield x
- notify("collectreport", rep)
+ self.config.api.pytest_collectreport(rep=rep)
if self.shouldstop:
break
@@ -81,12 +80,12 @@
""" setup any neccessary resources ahead of the test run. """
self.bus.notify("testrunstart")
- def pyevent__itemtestreport(self, rep):
+ def pytest_itemtestreport(self, rep):
if rep.failed:
self._testsfailed = True
if self.config.option.exitfirst:
self.shouldstop = True
- pyevent__collectreport = pyevent__itemtestreport
+ pytest_collectreport = pytest_itemtestreport
def sessionfinishes(self, exitstatus=0, excinfo=None):
""" teardown any resources after a test run. """
Modified: py/trunk/py/test/testing/test_collect.py
==============================================================================
--- py/trunk/py/test/testing/test_collect.py (original)
+++ py/trunk/py/test/testing/test_collect.py Thu Apr 9 01:33:48 2009
@@ -167,7 +167,7 @@
assert "hello" in wascalled
assert "world" in wascalled
# make sure the directories do not get double-appended
- colreports = sorter.getreports(names="collectreport")
+ colreports = sorter.getreports("collectreport")
names = [rep.colitem.name for rep in colreports]
assert names.count("hello") == 1
Modified: py/trunk/py/test/testing/test_pytestplugin.py
==============================================================================
--- py/trunk/py/test/testing/test_pytestplugin.py (original)
+++ py/trunk/py/test/testing/test_pytestplugin.py Thu Apr 9 01:33:48 2009
@@ -69,12 +69,12 @@
assert plugins.getplugin("plug1").__class__.__name__ == "Plug1Plugin"
assert plugins.getplugin("plug2").__class__.__name__ == "Plug2Plugin"
- def test_consider_module_import_module(self, testdir, EventRecorder):
+ def test_consider_module_import_module(self, testdir):
mod = py.std.new.module("x")
mod.pytest_plugins = "pytest_a"
aplugin = testdir.makepyfile(pytest_a="""class APlugin: pass""")
plugins = PytestPlugins()
- sorter = EventRecorder(plugins)
+ sorter = testdir.geteventrecorder(plugins)
#syspath.prepend(aplugin.dirpath())
py.std.sys.path.insert(0, str(aplugin.dirpath()))
plugins.consider_module(mod)
Modified: py/trunk/py/test/testing/test_runner.py
==============================================================================
--- py/trunk/py/test/testing/test_runner.py (original)
+++ py/trunk/py/test/testing/test_runner.py Thu Apr 9 01:33:48 2009
@@ -27,7 +27,7 @@
setup = SetupState()
res = setup.do_setup(item)
assert not res
- rep = evrec.popcall("itemsetupreport").rep
+ rep = evrec.popcall(pytest_itemsetupreport).rep
assert rep.failed
assert not rep.skipped
assert rep.excrepr
@@ -46,10 +46,10 @@
setup = SetupState()
res = setup.do_setup(item)
assert res
- rep = evrec.popcall("itemsetupreport").rep
+ rep = evrec.popcall(pytest_itemsetupreport).rep
assert rep.passed
setup.do_teardown(item)
- rep = evrec.popcall("itemsetupreport").rep
+ rep = evrec.popcall(pytest_itemsetupreport).rep
assert rep.item == item
assert rep.failed
assert not rep.passed
@@ -67,7 +67,7 @@
evrec = testdir.geteventrecorder(item.config)
setup = SetupState()
setup.do_setup(item)
- rep = evrec.popcall("itemsetupreport").rep
+ rep = evrec.popcall(pytest_itemsetupreport).rep
assert not rep.failed
assert rep.skipped
assert rep.excrepr
@@ -78,7 +78,7 @@
evrec = testdir.geteventrecorder(item.config)
setup = SetupState()
setup.do_fixture_and_runtest(item)
- rep = evrec.popcall("itemtestreport").rep
+ rep = evrec.popcall(pytest_itemtestreport).rep
assert rep.passed
def test_runtest_fails(self, testdir):
@@ -86,7 +86,7 @@
evrec = testdir.geteventrecorder(item.config)
setup = SetupState()
setup.do_fixture_and_runtest(item)
- event = evrec.popcall("item_runtest_finished")
+ event = evrec.popcall(pytest_item_runtest_finished)
assert event.excinfo
Modified: py/trunk/py/test/testing/test_session.py
==============================================================================
--- py/trunk/py/test/testing/test_session.py (original)
+++ py/trunk/py/test/testing/test_session.py Thu Apr 9 01:33:48 2009
@@ -27,7 +27,7 @@
assert failed[2].colitem.name == "test_two"
itemstarted = sorter.getcalls("itemstart")
assert len(itemstarted) == 4
- colstarted = sorter.getcalls("collectionstart")
+ colstarted = sorter.getcalls("collectstart")
assert len(colstarted) == 1
col = colstarted[0].collector
assert isinstance(col, py.test.collect.Module)
@@ -199,10 +199,10 @@
)
sorter = testdir.inline_run('--collectonly', p.dirpath())
- itemstarted = sorter.getcalls("itemstart")
+ itemstarted = sorter.getcalls("pytest_itemstart")
assert len(itemstarted) == 3
assert not sorter.getreports("itemtestreport")
- started = sorter.getcalls("collectionstart")
+ started = sorter.getcalls("pytest_collectstart")
finished = sorter.getreports("collectreport")
assert len(started) == len(finished)
assert len(started) == 8
More information about the pytest-commit
mailing list