From 85635e12398c92d0ee55022cb5b39d28c2f8503e Mon Sep 17 00:00:00 2001 From: hpk Date: Fri, 3 Apr 2009 18:26:21 +0200 Subject: [PATCH] [svn r63575] killing yet more test "events" and redundant code --HG-- branch : trunk --- py/test/dist/testing/test_dsession.py | 4 +- py/test/event.py | 27 ++-------- py/test/looponfail/remote.py | 8 +-- py/test/plugin/api.py | 4 +- py/test/plugin/pytest_pytester.py | 2 +- py/test/plugin/pytest_resultdb.py | 32 +++-------- py/test/plugin/pytest_resultlog.py | 78 ++++++++++++--------------- py/test/plugin/pytest_terminal.py | 14 ++--- 8 files changed, 60 insertions(+), 109 deletions(-) diff --git a/py/test/dist/testing/test_dsession.py b/py/test/dist/testing/test_dsession.py index 8b6efa8cb..8954035ed 100644 --- a/py/test/dist/testing/test_dsession.py +++ b/py/test/dist/testing/test_dsession.py @@ -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)) diff --git a/py/test/event.py b/py/test/event.py index 1900c1907..09373519f 100644 --- a/py/test/event.py +++ b/py/test/event.py @@ -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) -# diff --git a/py/test/looponfail/remote.py b/py/test/looponfail/remote.py index 621dea16f..71faeb53b 100644 --- a/py/test/looponfail/remote.py +++ b/py/test/looponfail/remote.py @@ -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]) diff --git a/py/test/plugin/api.py b/py/test/plugin/api.py index ce63ed9d1..ea2078e97 100644 --- a/py/test/plugin/api.py +++ b/py/test/plugin/api.py @@ -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): diff --git a/py/test/plugin/pytest_pytester.py b/py/test/plugin/pytest_pytester.py index e3c78bfac..46cef21ec 100644 --- a/py/test/plugin/pytest_pytester.py +++ b/py/test/plugin/pytest_pytester.py @@ -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) diff --git a/py/test/plugin/pytest_resultdb.py b/py/test/plugin/pytest_resultdb.py index 05983d634..7fc502fc9 100644 --- a/py/test/plugin/pytest_resultdb.py +++ b/py/test/plugin/pytest_resultdb.py @@ -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() diff --git a/py/test/plugin/pytest_resultlog.py b/py/test/plugin/pytest_resultlog.py index a9a2b18b7..5438e0720 100644 --- a/py/test/plugin/pytest_resultlog.py +++ b/py/test/plugin/pytest_resultlog.py @@ -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() diff --git a/py/test/plugin/pytest_terminal.py b/py/test/plugin/pytest_terminal.py index b0524104c..ff0d9a094 100644 --- a/py/test/plugin/pytest_terminal.py +++ b/py/test/plugin/pytest_terminal.py @@ -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*",