[svn r63560] beginning to review/polish test events

* pyevent() now receives args and kwargs as simple arguments
* refactoring event handling in tests

--HG--
branch : trunk
This commit is contained in:
hpk 2009-04-03 12:57:34 +02:00
parent 3ab894cadc
commit 54cc936437
11 changed files with 70 additions and 32 deletions

View File

@ -9,7 +9,7 @@ class EventQueue:
self.queue = queue
bus.register(self)
def pyevent(self, eventname, *args, **kwargs):
def pyevent(self, eventname, args, kwargs):
self.queue.put((eventname, args, kwargs))
def geteventargs(self, eventname, timeout=2.0):
@ -48,7 +48,7 @@ class MySetup:
assert not self.node.channel.isclosed()
return self.node
def finalize(self):
def xfinalize(self):
if hasattr(self, 'node'):
gw = self.node.gateway
print "exiting:", gw
@ -56,7 +56,7 @@ class MySetup:
def pytest_funcarg__mysetup(pyfuncitem):
mysetup = MySetup(pyfuncitem)
pyfuncitem.addfinalizer(mysetup.finalize)
#pyfuncitem.addfinalizer(mysetup.finalize)
return mysetup
def pytest_funcarg__testdir(__call__, pyfuncitem):

View File

@ -22,10 +22,6 @@ class NOP(BaseEvent):
# Basic Live Reporting Events
# ----------------------------------------------------------------------
class TestrunStart(BaseEvent):
def __init__(self):
self.timestart = time.time()
class TestrunFinish(BaseEvent):
def __init__(self, exitstatus=0, excinfo=None):
self.exitstatus = exitstatus

View File

@ -55,7 +55,7 @@ class PluginHooks:
class Events:
# Events
def pyevent(self, eventname, *args, **kwargs):
def pyevent(self, eventname, args, kwargs):
""" generically called for each notification event. """
def pyevent__gateway_init(self, gateway):
@ -76,7 +76,7 @@ class Events:
def pyevent__internalerror(self, event):
""" called for internal errors. """
def pyevent__itemstart(self, item, node):
def pyevent__itemstart(self, item, node=None):
""" test item gets collected. """
def pyevent__itemtestreport(self, event):
@ -91,7 +91,7 @@ class Events:
def pyevent__collectionreport(self, event):
""" collector finished collecting. """
def pyevent__testrunstart(self, event):
def pyevent__testrunstart(self):
""" whole test run starts. """
def pyevent__testrunfinish(self, event):

View File

@ -16,7 +16,7 @@ class EventlogPlugin:
self.eventlogfile.close()
del self.eventlogfile
def pyevent(self, eventname, *args, **kwargs):
def pyevent(self, eventname, args, kwargs):
if hasattr(self, 'eventlogfile'):
f = self.eventlogfile
print >>f, eventname, args, kwargs
@ -36,6 +36,6 @@ def test_generic(plugintester):
""")
testdir.runpytest("--eventlog=event.log")
s = testdir.tmpdir.join("event.log").read()
assert s.find("TestrunStart") != -1
assert s.find("testrunstart") != -1
assert s.find("ItemTestReport") != -1
assert s.find("TestrunFinish") != -1

View File

@ -21,8 +21,8 @@ class ExecnetcleanupPlugin:
if self._gateways is not None:
self._gateways.remove(gateway)
def pyevent__testrunstart(self, event):
self.trace("testrunstart", event)
def pyevent__testrunstart(self):
self.trace("testrunstart")
self._gateways = []
def pyevent__testrunfinish(self, event):

View File

@ -3,8 +3,10 @@ pytes plugin for easing testing of pytest runs themselves.
"""
import py
import inspect
from py.__.test import event
from py.__.test.config import Config as pytestConfig
import api
class PytesterPlugin:
def pytest_funcarg__linecomp(self, pyfuncitem):
@ -257,6 +259,13 @@ class Event:
def __repr__(self):
return "<Event %r %r>" %(self.name, self.args)
class ParsedEvent:
def __init__(self, locals):
self.__dict__ = locals.copy()
def __repr__(self):
return "<Event %r>" %(self.__dict__,)
class EventRecorder(object):
def __init__(self, pyplugins, debug=False): # True):
self.events = []
@ -264,7 +273,7 @@ class EventRecorder(object):
self.debug = debug
pyplugins.register(self)
def pyevent(self, name, *args, **kwargs):
def pyevent(self, name, args, kwargs):
if name == "plugin_registered" and args == (self,):
return
if self.debug:
@ -278,12 +287,42 @@ class EventRecorder(object):
return event
raise KeyError("popevent: %r not found in %r" %(name, self.events))
def getevents(self, eventname):
method = self.geteventmethod(eventname)
l = []
for event in self.events:
if event.name == eventname:
pevent = method(*event.args, **event.kwargs)
l.append(pevent)
return l
def geteventmethod(self, eventname):
mname = "pyevent__" + eventname
method = getattr(api.Events, mname)
args, varargs, varkw, default = inspect.getargspec(method)
assert args[0] == "self"
args = args[1:]
fspec = inspect.formatargspec(args, varargs, varkw, default)
source = """def %(mname)s%(fspec)s:
return ParsedEvent(locals())""" % locals()
print source
exec source
return locals()[mname]
def firstparsedevent(self, eventname):
return self.parsedevents(eventname)[0]
def get(self, cls):
l = []
for event in self.events:
value = event.args[0]
if isinstance(value, cls):
l.append(value)
try:
value = event.args[0]
except IndexError:
continue
else:
if isinstance(value, cls):
l.append(value)
return l
def getnamed(self, *names):

View File

@ -370,7 +370,7 @@ class TestWithFunctionIntegration:
except ValueError:
excinfo = event.InternalException()
reslog = ResultDB(StringIO.StringIO())
reslog.pyevent("internalerror", excinfo)
reslog.pyevent("internalerror", (excinfo,), {})
entry = reslog.logfile.getvalue()
entry_lines = entry.splitlines()

View File

@ -81,7 +81,9 @@ class ResultLog(object):
shortrepr, longrepr = getoutcomecodes(event)
self.write_log_entry(shortrepr, gpath, longrepr)
def pyevent(self, eventname, event, *args, **kwargs):
def pyevent(self, eventname, args, kwargs):
if args:
event = args[0]
if eventname == "itemtestreport":
self.log_outcome(event)
elif eventname == "collectionreport":
@ -91,6 +93,7 @@ class ResultLog(object):
path = event.repr.reprcrash.path # fishing :(
self.write_log_entry('!', path, str(event.repr))
# ===============================================================================
#
# plugin tests
@ -223,7 +226,7 @@ class TestWithFunctionIntegration:
except ValueError:
excinfo = event.InternalException()
reslog = ResultLog(StringIO.StringIO())
reslog.pyevent("internalerror", excinfo)
reslog.pyevent("internalerror", (excinfo,), {})
entry = reslog.logfile.getvalue()
entry_lines = entry.splitlines()

View File

@ -194,7 +194,7 @@ class TerminalReporter:
self.stats.setdefault("skipped", []).append(event)
self.write_fspath_result(event.colitem.fspath, "S")
def pyevent__testrunstart(self, event):
def pyevent__testrunstart(self):
self.write_sep("=", "test session starts", bold=True)
self._sessionstarttime = py.std.time.time()
@ -370,7 +370,7 @@ class TestTerminal:
""")
rep = TerminalReporter(modcol.config, file=linecomp.stringio)
rep.config.bus.register(rep)
rep.config.bus.notify("testrunstart", event.TestrunStart())
rep.config.bus.notify("testrunstart")
for item in testdir.genitems([modcol]):
ev = basic_run_report(item)
@ -397,7 +397,7 @@ class TestTerminal:
""", configargs=("-v",))
rep = TerminalReporter(modcol.config, file=linecomp.stringio)
rep.config.bus.register(rep)
rep.config.bus.notify("testrunstart", event.TestrunStart())
rep.config.bus.notify("testrunstart")
items = modcol.collect()
rep.config.option.debug = True #
for item in items:
@ -422,7 +422,7 @@ class TestTerminal:
modcol = testdir.getmodulecol("import xyz")
rep = TerminalReporter(modcol.config, file=linecomp.stringio)
rep.config.bus.register(rep)
rep.config.bus.notify("testrunstart", event.TestrunStart())
rep.config.bus.notify("testrunstart")
l = list(testdir.genitems([modcol]))
assert len(l) == 0
linecomp.assert_contains_lines([
@ -518,8 +518,8 @@ class TestTerminal:
""", configargs=("--tb=%s" % tbopt,))
rep = TerminalReporter(modcol.config, file=linecomp.stringio)
rep.config.bus.register(rep)
rep.config.bus.notify("testrunstart", event.TestrunStart())
rep.config.bus.notify("testrunstart", event.TestrunStart())
rep.config.bus.notify("testrunstart")
rep.config.bus.notify("testrunstart")
for item in testdir.genitems([modcol]):
rep.config.bus.notify("itemtestreport", basic_run_report(item))
rep.config.bus.notify("testrunfinish", event.TestrunFinish())
@ -566,7 +566,7 @@ class TestTerminal:
rep = TerminalReporter(modcol.config, file=linecomp.stringio)
modcol.config.bus.register(rep)
bus = modcol.config.bus
bus.notify("testrunstart", event.TestrunStart())
bus.notify("testrunstart")
try:
for item in testdir.genitems([modcol]):
bus.notify("itemtestreport", basic_run_report(item))

View File

@ -79,7 +79,7 @@ class Session(object):
def sessionstarts(self):
""" setup any neccessary resources ahead of the test run. """
self.bus.notify("testrunstart", event.TestrunStart())
self.bus.notify("testrunstart")
def pyevent__itemtestreport(self, rep):
if rep.failed:

View File

@ -25,11 +25,11 @@ class SessionTests:
assert failed[0].colitem.name == "test_one_one"
assert failed[1].colitem.name == "test_other"
assert failed[2].colitem.name == "test_two"
itemstarted = sorter.getnamed("itemstart")
itemstarted = sorter.getevents("itemstart")
assert len(itemstarted) == 4
colstarted = sorter.getnamed("collectionstart")
colstarted = sorter.getevents("collectionstart")
assert len(colstarted) == 1
col = colstarted[0].collector
col = colstarted[0].event.collector
assert isinstance(col, py.test.collect.Module)
def test_nested_import_error(self, testdir):