[py-svn] r63575 - in py/trunk/py/test: . dist/testing looponfail plugin
hpk at codespeak.net
hpk at codespeak.net
Fri Apr 3 18:26:22 CEST 2009
Author: hpk
Date: Fri Apr 3 18:26:21 2009
New Revision: 63575
Modified:
py/trunk/py/test/dist/testing/test_dsession.py
py/trunk/py/test/event.py
py/trunk/py/test/looponfail/remote.py
py/trunk/py/test/plugin/api.py
py/trunk/py/test/plugin/pytest_pytester.py
py/trunk/py/test/plugin/pytest_resultdb.py
py/trunk/py/test/plugin/pytest_resultlog.py
py/trunk/py/test/plugin/pytest_terminal.py
Log:
killing yet more test "events" and redundant code
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 Fri Apr 3 18:26:21 2009
@@ -131,9 +131,9 @@
# check that RescheduleEvents are not immediately
# rescheduled if there are no nodes
assert loopstate.dowork == False
- session.queueevent("anonymous", event.NOP())
+ session.queueevent("anonymous")
session.loop_once(loopstate)
- session.queueevent("anonymous", event.NOP())
+ session.queueevent("anonymous")
session.loop_once(loopstate)
assert node.sent == [[item]]
session.queueevent("itemtestreport", run(item, node))
Modified: py/trunk/py/test/event.py
==============================================================================
--- py/trunk/py/test/event.py (original)
+++ py/trunk/py/test/event.py Fri Apr 3 18:26:21 2009
@@ -6,20 +6,16 @@
import time
from py.__.test.outcome import Skipped
-class BaseEvent(object):
+# ----------------------------------------------------------------------
+# Events related to collecting and executing test Items
+# ----------------------------------------------------------------------
+
+class BaseReport(object):
def __repr__(self):
l = ["%s=%s" %(key, value)
for key, value in self.__dict__.items()]
return "<%s %s>" %(self.__class__.__name__, " ".join(l),)
-class NOP(BaseEvent):
- pass
-
-# ----------------------------------------------------------------------
-# Events related to collecting and executing test Items
-# ----------------------------------------------------------------------
-
-class BaseReport(BaseEvent):
def toterminal(self, out):
longrepr = self.longrepr
if hasattr(longrepr, 'toterminal'):
@@ -93,16 +89,3 @@
else:
out.line(str(longrepr))
-class LooponfailingInfo(BaseEvent):
- def __init__(self, failreports, rootdirs):
- self.failreports = failreports
- self.rootdirs = rootdirs
-
-# make all eventclasses available on BaseEvent so that
-# consumers of events can easily filter by
-# 'isinstance(event, event.Name)' checks
-
-for name, cls in vars().items():
- if hasattr(cls, '__bases__') and issubclass(cls, BaseEvent):
- setattr(BaseEvent, name, cls)
-#
Modified: py/trunk/py/test/looponfail/remote.py
==============================================================================
--- py/trunk/py/test/looponfail/remote.py (original)
+++ py/trunk/py/test/looponfail/remote.py Fri Apr 3 18:26:21 2009
@@ -39,7 +39,6 @@
colitems = loopstate.colitems
loopstate.wasfailing = colitems and len(colitems)
loopstate.colitems = self.remotecontrol.runsession(colitems or ())
- #ev = event.LooponfailingInfo(loopstate.failreports, self.rootdirs)
self.remotecontrol.setup()
class LoopState:
@@ -149,6 +148,7 @@
DEBUG("SLAVE: starting session.main()")
session.main(colitems)
- ev = event.LooponfailingInfo(list(failreports), [config.topdir])
- session.bus.notify("looponfailinfo", ev)
- channel.send([x.colitem._totrail() for x in failreports if x.failed])
+ session.bus.notify("looponfailinfo",
+ failreports=list(failreports),
+ rootdirs=[config.topdir])
+ channel.send([x.colitem._totrail() for x in failreports])
Modified: py/trunk/py/test/plugin/api.py
==============================================================================
--- py/trunk/py/test/plugin/api.py (original)
+++ py/trunk/py/test/plugin/api.py Fri Apr 3 18:26:21 2009
@@ -108,10 +108,10 @@
def pyevent__testnodedown(self, node, error):
""" Test Node is down. """
- def pyevent__rescheduleitems(self, event):
+ def pyevent__rescheduleitems(self, items):
""" reschedule Items from a node that went down. """
- def pyevent__looponfailinfo(self, event):
+ def pyevent__looponfailinfo(self, failreports, rootdirs):
""" info for repeating failing tests. """
def pyevent__plugin_registered(self, plugin):
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 Fri Apr 3 18:26:21 2009
@@ -394,7 +394,7 @@
def test_eventrecorder():
bus = py._com.PyPlugins()
recorder = EventRecorder(bus)
- bus.notify("anonymous", event.NOP())
+ bus.notify("anonymous")
assert recorder.events
assert not recorder.getfailures()
rep = event.ItemTestReport(None, None)
Modified: py/trunk/py/test/plugin/pytest_resultdb.py
==============================================================================
--- py/trunk/py/test/plugin/pytest_resultdb.py (original)
+++ py/trunk/py/test/plugin/pytest_resultdb.py Fri Apr 3 18:26:21 2009
@@ -1,9 +1,9 @@
import uuid
import py
-from pytest_resultlog import generic_path, getoutcomecodes
+from pytest_resultlog import ResultLog
class ResultdbPlugin:
- """resultdb plugin for database logging of test results.
+ """XXX in progress: resultdb plugin for database logging of test results.
Saves test results to a datastore.
@@ -144,12 +144,12 @@
return d
-class ResultDB(object):
+class ResultDB(ResultLog):
def __init__(self, cls, db_path):
self.archive = cls(db_path)
self.archive.init_db()
- def write_log_entry(self, event, shortrepr, name, longrepr):
+ def write_log_entry(self, testpath, shortrepr, longrepr):
data = {}
event_excludes = ['colitem', 'longrepr']
for item in vars(event).keys():
@@ -159,29 +159,9 @@
data['longrepr'] = longrepr
data['shortrepr'] = shortrepr
- data['fspath'] = unicode(event.colitem.fspath)
- data['itemname'] = event.colitem.name
-
- data['name'] = name
+ data['testpath'] = unicode(testpath)
self.archive.append_data([data])
- def log_outcome(self, event):
- if (not event.passed or isinstance(event, event.ItemTestReport)):
- gpath = generic_path(event.colitem)
- shortrepr, longrepr = getoutcomecodes(event)
- self.write_log_entry(event, shortrepr, gpath, longrepr)
-
- def pyevent__itemtestreport(self, event):
- self.log_outcome(event)
-
- def pyevent__collectionreport(self, event):
- if not event.passed:
- self.log_outcome(event)
-
- def pyevent__internalerror(self, excrepr):
- path = excrepr.reprcrash.path
- XXX # we don't have an event
- self.write_log_entry(event, '!', path, str(excrepr))
SQL_CREATE_TABLES = """
create table pytest_results (
@@ -370,7 +350,7 @@
except ValueError:
excinfo = py.code.ExceptionInfo()
reslog = ResultDB(StringIO.StringIO())
- reslog.pyevent("internalerror", (excinfo.getrepr(),), {})
+ reslog.pyevent__internalerror(excinfo.getrepr)
entry = reslog.logfile.getvalue()
entry_lines = entry.splitlines()
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 Fri Apr 3 18:26:21 2009
@@ -44,55 +44,43 @@
gpath.append(name)
fspath = newfspath
return ''.join(gpath)
-
-def getoutcomecodes(ev):
- if isinstance(ev, ev.CollectionReport):
- # encode pass/fail/skip indepedent of terminal reporting semantics
- # XXX handle collection and item reports more uniformly
- assert not ev.passed
- if ev.failed:
- code = "F"
- elif ev.skipped:
- code = "S"
- longrepr = str(ev.longrepr.reprcrash)
- else:
- assert isinstance(ev, ev.ItemTestReport)
- code = ev.shortrepr
- if ev.passed:
- longrepr = ""
- elif ev.failed:
- longrepr = str(ev.longrepr)
- elif ev.skipped:
- longrepr = str(ev.longrepr.reprcrash.message)
- return code, longrepr
class ResultLog(object):
def __init__(self, logfile):
self.logfile = logfile # preferably line buffered
- def write_log_entry(self, shortrepr, name, longrepr):
- print >>self.logfile, "%s %s" % (shortrepr, name)
+ def write_log_entry(self, testpath, shortrepr, longrepr):
+ print >>self.logfile, "%s %s" % (shortrepr, testpath)
for line in longrepr.splitlines():
print >>self.logfile, " %s" % line
- def log_outcome(self, event):
- if (not event.passed or isinstance(event, event.ItemTestReport)):
- gpath = generic_path(event.colitem)
- shortrepr, longrepr = getoutcomecodes(event)
- self.write_log_entry(shortrepr, gpath, longrepr)
-
- def pyevent(self, eventname, args, kwargs):
- if args:
- event = args[0]
- if eventname == "itemtestreport":
- self.log_outcome(event)
- elif eventname == "collectionreport":
- if not event.passed:
- self.log_outcome(event)
- elif eventname == "internalerror":
- excrepr = args[0]
- path = excrepr.reprcrash.path # fishing :(
- self.write_log_entry('!', path, str(excrepr))
+ def log_outcome(self, event, shortrepr, longrepr):
+ testpath = generic_path(event.colitem)
+ self.write_log_entry(testpath, shortrepr, longrepr)
+
+ def pyevent__itemtestreport(self, event):
+ code = event.shortrepr
+ if event.passed:
+ longrepr = ""
+ elif event.failed:
+ longrepr = str(event.longrepr)
+ elif event.skipped:
+ longrepr = str(event.longrepr.reprcrash.message)
+ self.log_outcome(event, code, longrepr)
+
+ def pyevent__collectionreport(self, event):
+ if not event.passed:
+ if event.failed:
+ code = "F"
+ else:
+ assert event.skipped
+ code = "S"
+ longrepr = str(event.longrepr.reprcrash)
+ self.log_outcome(event, code, longrepr)
+
+ def pyevent__internalerror(self, excrepr):
+ path = excrepr.reprcrash.path
+ self.write_log_entry(path, '!', str(excrepr))
# ===============================================================================
@@ -127,7 +115,7 @@
def test_write_log_entry():
reslog = ResultLog(None)
reslog.logfile = StringIO.StringIO()
- reslog.write_log_entry('.', 'name', '')
+ reslog.write_log_entry('name', '.', '')
entry = reslog.logfile.getvalue()
assert entry[-1] == '\n'
entry_lines = entry.splitlines()
@@ -135,7 +123,7 @@
assert entry_lines[0] == '. name'
reslog.logfile = StringIO.StringIO()
- reslog.write_log_entry('s', 'name', 'Skipped')
+ reslog.write_log_entry('name', 's', 'Skipped')
entry = reslog.logfile.getvalue()
assert entry[-1] == '\n'
entry_lines = entry.splitlines()
@@ -144,7 +132,7 @@
assert entry_lines[1] == ' Skipped'
reslog.logfile = StringIO.StringIO()
- reslog.write_log_entry('s', 'name', 'Skipped\n')
+ reslog.write_log_entry('name', 's', 'Skipped\n')
entry = reslog.logfile.getvalue()
assert entry[-1] == '\n'
entry_lines = entry.splitlines()
@@ -154,7 +142,7 @@
reslog.logfile = StringIO.StringIO()
longrepr = ' tb1\n tb 2\nE tb3\nSome Error'
- reslog.write_log_entry('F', 'name', longrepr)
+ reslog.write_log_entry('name', 'F', longrepr)
entry = reslog.logfile.getvalue()
assert entry[-1] == '\n'
entry_lines = entry.splitlines()
@@ -227,7 +215,7 @@
except ValueError:
excinfo = py.code.ExceptionInfo()
reslog = ResultLog(StringIO.StringIO())
- reslog.pyevent("internalerror", (excinfo.getrepr(),), {})
+ reslog.pyevent__internalerror(excinfo.getrepr())
entry = reslog.logfile.getvalue()
entry_lines = entry.splitlines()
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 Fri Apr 3 18:26:21 2009
@@ -154,9 +154,9 @@
#self.write_fspath_result(fspath, "")
self.write_ensure_prefix(line, "")
- def pyevent__rescheduleitems(self, event):
+ def pyevent__rescheduleitems(self, items):
if self.config.option.debug:
- self.write_sep("!", "RESCHEDULING %s " %(event.items,))
+ self.write_sep("!", "RESCHEDULING %s " %(items,))
def pyevent__deselected(self, items):
self.stats.setdefault('deselected', []).append(items)
@@ -232,17 +232,17 @@
self.summary_deselected()
self.summary_stats()
- def pyevent__looponfailinfo(self, event):
- if event.failreports:
+ def pyevent__looponfailinfo(self, failreports, rootdirs):
+ if failreports:
self.write_sep("#", "LOOPONFAILING", red=True)
- for report in event.failreports:
+ for report in failreports:
try:
loc = report.longrepr.reprcrash
except AttributeError:
loc = str(report.longrepr)[:50]
self.write_line(loc, red=True)
self.write_sep("#", "waiting for changes")
- for rootdir in event.rootdirs:
+ for rootdir in rootdirs:
self.write_line("### Watching: %s" %(rootdir,), bold=True)
#
@@ -496,7 +496,7 @@
""")
rep = TerminalReporter(modcol.config, file=linecomp.stringio)
reports = [basic_run_report(x) for x in modcol.collect()]
- rep.pyevent__looponfailinfo(event.LooponfailingInfo(reports, [modcol.config.topdir]))
+ rep.pyevent__looponfailinfo(reports, [modcol.config.topdir])
linecomp.assert_contains_lines([
"*test_looponfailreport.py:2: assert 0",
"*test_looponfailreport.py:4: ValueError*",
More information about the pytest-commit
mailing list