From 4aeb929b3c5e5030c903c4b7adfd694af4100baa Mon Sep 17 00:00:00 2001 From: hpk Date: Sat, 4 Apr 2009 02:34:20 +0200 Subject: [PATCH] [svn r63600] merge remaining content of event.py into runner.py. --HG-- branch : trunk --- py/test/dist/dsession.py | 5 +- py/test/dist/testing/test_dsession.py | 1 - py/test/dist/testing/test_nodemanage.py | 2 - py/test/event.py | 90 ------------------------- py/test/plugin/pytest_default.py | 4 +- py/test/plugin/pytest_pytester.py | 12 ++-- py/test/plugin/pytest_resultdb.py | 1 - py/test/plugin/pytest_resultlog.py | 1 - py/test/plugin/pytest_terminal.py | 21 +++--- py/test/runner.py | 89 ++++++++++++++++++++++-- 10 files changed, 104 insertions(+), 122 deletions(-) delete mode 100644 py/test/event.py diff --git a/py/test/dist/dsession.py b/py/test/dist/dsession.py index 7e8ffa301..1cde5d88e 100644 --- a/py/test/dist/dsession.py +++ b/py/test/dist/dsession.py @@ -5,7 +5,7 @@ """ import py -from py.__.test.runner import basic_run_report, basic_collect_report +from py.__.test.runner import basic_run_report, basic_collect_report, ItemTestReport from py.__.test.session import Session from py.__.test import outcome from py.__.test.dist.nodemanage import NodeManager @@ -236,9 +236,8 @@ class DSession(Session): self.node2pending[node].remove(item) def handle_crashitem(self, item, node): - from py.__.test import event longrepr = "!!! Node %r crashed during running of test %r" %(node, item) - rep = event.ItemTestReport(item, when="???", excinfo=longrepr) + rep = ItemTestReport(item, when="???", excinfo=longrepr) rep.node = node self.bus.notify("itemtestreport", rep) diff --git a/py/test/dist/testing/test_dsession.py b/py/test/dist/testing/test_dsession.py index bf351c783..903561192 100644 --- a/py/test/dist/testing/test_dsession.py +++ b/py/test/dist/testing/test_dsession.py @@ -1,6 +1,5 @@ from py.__.test.dist.dsession import DSession from py.__.test.runner import basic_collect_report -from py.__.test import event from py.__.test import outcome import py diff --git a/py/test/dist/testing/test_nodemanage.py b/py/test/dist/testing/test_nodemanage.py index bbb698e5b..8cb801032 100644 --- a/py/test/dist/testing/test_nodemanage.py +++ b/py/test/dist/testing/test_nodemanage.py @@ -5,8 +5,6 @@ import py from py.__.test.dist.nodemanage import NodeManager -from py.__.test import event - def pytest_funcarg__source(pyfuncitem): return py.test.ensuretemp(pyfuncitem.getmodpath()).mkdir("source") def pytest_funcarg__dest(pyfuncitem): diff --git a/py/test/event.py b/py/test/event.py deleted file mode 100644 index 9cbacb250..000000000 --- a/py/test/event.py +++ /dev/null @@ -1,90 +0,0 @@ -""" - test collection and execution events -""" - -import py -import time -from py.__.test.outcome import Skipped - -# ---------------------------------------------------------------------- -# 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),) - - def toterminal(self, out): - longrepr = self.longrepr - if hasattr(longrepr, 'toterminal'): - longrepr.toterminal(out) - else: - out.line(str(longrepr)) - -class ItemTestReport(BaseReport): - """ Test Execution Report. """ - failed = passed = skipped = False - - def __init__(self, colitem, excinfo=None, when=None, outerr=None): - self.colitem = colitem - if colitem and when != "setup": - self.keywords = colitem.readkeywords() - else: - # if we fail during setup it might mean - # we are not able to access the underlying object - # this might e.g. happen if we are unpickled - # and our parent collector did not collect us - # (because it e.g. skipped for platform reasons) - self.keywords = {} - if not excinfo: - self.passed = True - self.shortrepr = "." - else: - self.when = when - if not isinstance(excinfo, py.code.ExceptionInfo): - self.failed = True - shortrepr = "?" - longrepr = excinfo - elif excinfo.errisinstance(Skipped): - self.skipped = True - shortrepr = "s" - longrepr = self.colitem._repr_failure_py(excinfo, outerr) - else: - self.failed = True - shortrepr = self.colitem.shortfailurerepr - if self.when == "execute": - longrepr = self.colitem.repr_failure(excinfo, outerr) - else: # exception in setup or teardown - longrepr = self.colitem._repr_failure_py(excinfo, outerr) - shortrepr = shortrepr.lower() - self.shortrepr = shortrepr - self.longrepr = longrepr - - -class CollectionReport(BaseReport): - """ Collection Report. """ - skipped = failed = passed = False - - def __init__(self, colitem, result, excinfo=None, outerr=None): - self.colitem = colitem - if not excinfo: - self.passed = True - self.result = result - else: - self.outerr = outerr - self.longrepr = self.colitem._repr_failure_py(excinfo, outerr) - if excinfo.errisinstance(Skipped): - self.skipped = True - self.reason = str(excinfo.value) - else: - self.failed = True - - def toterminal(self, out): - longrepr = self.longrepr - if hasattr(longrepr, 'toterminal'): - longrepr.toterminal(out) - else: - out.line(str(longrepr)) - diff --git a/py/test/plugin/pytest_default.py b/py/test/plugin/pytest_default.py index d6a07552c..94890f489 100644 --- a/py/test/plugin/pytest_default.py +++ b/py/test/plugin/pytest_default.py @@ -14,8 +14,8 @@ class DefaultPlugin: return True def pytest_item_makereport(self, item, excinfo, when, outerr): - from py.__.test import event - return event.ItemTestReport(item, excinfo, when, outerr) + from py.__.test import runner + return runner.ItemTestReport(item, excinfo, when, outerr) def pytest_pyfunc_call(self, pyfuncitem, args, kwargs): pyfuncitem.obj(*args, **kwargs) diff --git a/py/test/plugin/pytest_pytester.py b/py/test/plugin/pytest_pytester.py index 46cef21ec..b728a384a 100644 --- a/py/test/plugin/pytest_pytester.py +++ b/py/test/plugin/pytest_pytester.py @@ -4,7 +4,7 @@ pytes plugin for easing testing of pytest runs themselves. import py import inspect -from py.__.test import event +from py.__.test import runner from py.__.test.config import Config as pytestConfig import api @@ -374,7 +374,7 @@ class EventRecorder(object): """ return a testreport whose dotted import path matches """ __tracebackhide__ = True l = [] - for rep in self.get(event.ItemTestReport): + for rep in self.get(runner.ItemTestReport): if inamepart in rep.colitem.listnames(): l.append(rep) if not l: @@ -397,23 +397,23 @@ def test_eventrecorder(): bus.notify("anonymous") assert recorder.events assert not recorder.getfailures() - rep = event.ItemTestReport(None, None) + rep = runner.ItemTestReport(None, None) rep.passed = False rep.failed = True bus.notify("itemtestreport", rep) failures = recorder.getfailures() assert failures == [rep] - failures = recorder.get(event.ItemTestReport) + failures = recorder.get(runner.ItemTestReport) assert failures == [rep] failures = recorder.getnamed("itemtestreport") assert failures == [rep] - rep = event.ItemTestReport(None, None) + rep = runner.ItemTestReport(None, None) rep.passed = False rep.skipped = True bus.notify("itemtestreport", rep) - rep = event.CollectionReport(None, None) + rep = runner.CollectionReport(None, None) rep.passed = False rep.failed = True bus.notify("itemtestreport", rep) diff --git a/py/test/plugin/pytest_resultdb.py b/py/test/plugin/pytest_resultdb.py index 7fc502fc9..8190932cb 100644 --- a/py/test/plugin/pytest_resultdb.py +++ b/py/test/plugin/pytest_resultdb.py @@ -344,7 +344,6 @@ class TestWithFunctionIntegration: py.test.skip("Needs a rewrite for db version.") # they are produced for example by a teardown failing # at the end of the run - from py.__.test import event try: raise ValueError except ValueError: diff --git a/py/test/plugin/pytest_resultlog.py b/py/test/plugin/pytest_resultlog.py index 5438e0720..55103375d 100644 --- a/py/test/plugin/pytest_resultlog.py +++ b/py/test/plugin/pytest_resultlog.py @@ -209,7 +209,6 @@ class TestWithFunctionIntegration: def test_internal_exception(self): # they are produced for example by a teardown failing # at the end of the run - from py.__.test import event try: raise ValueError except ValueError: diff --git a/py/test/plugin/pytest_terminal.py b/py/test/plugin/pytest_terminal.py index dfdb9085d..b58f94525 100644 --- a/py/test/plugin/pytest_terminal.py +++ b/py/test/plugin/pytest_terminal.py @@ -353,8 +353,7 @@ def repr_pythonversion(v=None): # # =============================================================================== -from py.__.test import event -from py.__.test.runner import basic_run_report +from py.__.test import runner class TestTerminal: @@ -373,7 +372,7 @@ class TestTerminal: rep.config.bus.notify("testrunstart") for item in testdir.genitems([modcol]): - ev = basic_run_report(item) + ev = runner.basic_run_report(item) rep.config.bus.notify("itemtestreport", ev) linecomp.assert_contains_lines([ "*test_pass_skip_fail.py .sF" @@ -404,7 +403,7 @@ class TestTerminal: rep.config.bus.notify("itemstart", item, None) s = linecomp.stringio.getvalue().strip() assert s.endswith(item.name) - rep.config.bus.notify("itemtestreport", basic_run_report(item)) + rep.config.bus.notify("itemtestreport", runner.basic_run_report(item)) linecomp.assert_contains_lines([ "*test_pass_skip_fail_verbose.py:2: *test_ok*PASS*", @@ -495,7 +494,7 @@ class TestTerminal: raise ValueError() """) rep = TerminalReporter(modcol.config, file=linecomp.stringio) - reports = [basic_run_report(x) for x in modcol.collect()] + reports = [runner.basic_run_report(x) for x in modcol.collect()] rep.pyevent__looponfailinfo(reports, [modcol.config.topdir]) linecomp.assert_contains_lines([ "*test_looponfailreport.py:2: assert 0", @@ -521,7 +520,8 @@ class TestTerminal: 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("itemtestreport", + runner.basic_run_report(item)) rep.config.bus.notify("testrunfinish", exitstatus=1) s = linecomp.stringio.getvalue() if tbopt == "long": @@ -569,7 +569,8 @@ class TestTerminal: bus.notify("testrunstart") try: for item in testdir.genitems([modcol]): - bus.notify("itemtestreport", basic_run_report(item)) + bus.notify("itemtestreport", + runner.basic_run_report(item)) except KeyboardInterrupt: excinfo = py.code.ExceptionInfo() else: @@ -602,12 +603,12 @@ class TestTerminal: lineno = 3 message = "justso" - ev1 = event.CollectionReport(None, None) + ev1 = runner.CollectionReport(None, None) ev1.when = "execute" ev1.skipped = True ev1.longrepr = longrepr - ev2 = event.ItemTestReport(None, excinfo=longrepr) + ev2 = runner.ItemTestReport(None, excinfo=longrepr) ev2.skipped = True l = folded_skips([ev1, ev2]) @@ -637,7 +638,7 @@ class TestCollectonly: " ", ]) rep.config.bus.notify( "collectionreport", - event.CollectionReport(modcol, [], excinfo=None)) + runner.CollectionReport(modcol, [], excinfo=None)) assert rep.indent == indent def test_collectonly_skipped_module(self, testdir, linecomp): diff --git a/py/test/runner.py b/py/test/runner.py index 59782ef66..6810eda1c 100644 --- a/py/test/runner.py +++ b/py/test/runner.py @@ -8,9 +8,7 @@ import py, os, sys -from py.__.test import event -from py.__.test.outcome import Exit -from py.__.test.dist.mypickle import ImmutablePickler +from py.__.test.outcome import Exit, Skipped def basic_run_report(item, pdb=None): """ return report about setting up and running a test item. """ @@ -54,14 +52,14 @@ def basic_collect_report(collector): raise except: excinfo = py.code.ExceptionInfo() - return event.CollectionReport(collector, res, excinfo, outerr) + return CollectionReport(collector, res, excinfo, outerr) from cPickle import Pickler, Unpickler from cStringIO import StringIO def forked_run_report(item, pdb=None): EXITSTATUS_TESTEXIT = 4 - + from py.__.test.dist.mypickle import ImmutablePickler ipickle = ImmutablePickler(uneven=0) ipickle.selfmemoize(item.config) def runforked(): @@ -86,4 +84,83 @@ def report_process_crash(item, result): ("X", "CRASHED"), ("%s:%s: CRASHED with signal %d" %(path, lineno, result.signal)), ] - return event.ItemTestReport(item, excinfo=longrepr, when="???") + return ItemTestReport(item, excinfo=longrepr, when="???") + +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),) + + def toterminal(self, out): + longrepr = self.longrepr + if hasattr(longrepr, 'toterminal'): + longrepr.toterminal(out) + else: + out.line(str(longrepr)) + +class ItemTestReport(BaseReport): + """ Test Execution Report. """ + failed = passed = skipped = False + + def __init__(self, colitem, excinfo=None, when=None, outerr=None): + self.colitem = colitem + if colitem and when != "setup": + self.keywords = colitem.readkeywords() + else: + # if we fail during setup it might mean + # we are not able to access the underlying object + # this might e.g. happen if we are unpickled + # and our parent collector did not collect us + # (because it e.g. skipped for platform reasons) + self.keywords = {} + if not excinfo: + self.passed = True + self.shortrepr = "." + else: + self.when = when + if not isinstance(excinfo, py.code.ExceptionInfo): + self.failed = True + shortrepr = "?" + longrepr = excinfo + elif excinfo.errisinstance(Skipped): + self.skipped = True + shortrepr = "s" + longrepr = self.colitem._repr_failure_py(excinfo, outerr) + else: + self.failed = True + shortrepr = self.colitem.shortfailurerepr + if self.when == "execute": + longrepr = self.colitem.repr_failure(excinfo, outerr) + else: # exception in setup or teardown + longrepr = self.colitem._repr_failure_py(excinfo, outerr) + shortrepr = shortrepr.lower() + self.shortrepr = shortrepr + self.longrepr = longrepr + + +class CollectionReport(BaseReport): + """ Collection Report. """ + skipped = failed = passed = False + + def __init__(self, colitem, result, excinfo=None, outerr=None): + self.colitem = colitem + if not excinfo: + self.passed = True + self.result = result + else: + self.outerr = outerr + self.longrepr = self.colitem._repr_failure_py(excinfo, outerr) + if excinfo.errisinstance(Skipped): + self.skipped = True + self.reason = str(excinfo.value) + else: + self.failed = True + + def toterminal(self, out): + longrepr = self.longrepr + if hasattr(longrepr, 'toterminal'): + longrepr.toterminal(out) + else: + out.line(str(longrepr)) +