p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[svn r63575] killing yet more test "events" and redundant code 

--HG--
branch : trunk
This commit is contained in:
hpk 2009-04-03 18:26:21 +02:00
parent 53491b1531
commit 85635e1239
8 changed files with 60 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
py/test/dist/testing/test_dsession.py vendored
View File

@ -131,9 +131,9 @@ class TestDSession:
# 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))

27
py/test/event.py
View File

@ -6,20 +6,16 @@ import py
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 @@ class CollectionReport(BaseReport):
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)
#

8
py/test/looponfail/remote.py
View File

@ -39,7 +39,6 @@ class LooponfailingSession(Session):
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 @@ def slave_runsession(channel, config, fullwidth, hasmarkup):
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])

4
py/test/plugin/api.py
View File

@ -108,10 +108,10 @@ class Events:
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):

2
py/test/plugin/pytest_pytester.py
View File

@ -394,7 +394,7 @@ class EventRecorder(object):
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)

32
py/test/plugin/pytest_resultdb.py
View File

@ -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 @@ class SQLiteResultArchive(object):
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 @@ class ResultDB(object):
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 @@ class TestWithFunctionIntegration:
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()

78
py/test/plugin/pytest_resultlog.py
View File

@ -44,55 +44,43 @@ def generic_path(item):
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 log_outcome(self, event, shortrepr, longrepr):
testpath = generic_path(event.colitem)
self.write_log_entry(testpath, shortrepr, 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 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_generic_path():
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 @@ def test_write_log_entry():
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 @@ def test_write_log_entry():
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 @@ def test_write_log_entry():
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 @@ class TestWithFunctionIntegration:
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()

14
py/test/plugin/pytest_terminal.py
View File

@ -154,9 +154,9 @@ class TerminalReporter:
#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 @@ class TerminalReporter:
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 @@ class TestTerminal:
""")
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*",