[svn r63600] merge remaining content of event.py into runner.py.

--HG--
branch : trunk
This commit is contained in:
hpk 2009-04-04 02:34:20 +02:00
parent 7b47da2203
commit 4aeb929b3c
10 changed files with 104 additions and 122 deletions

View File

@ -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)

View File

@ -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

View File

@ -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):

View File

@ -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))

View File

@ -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)

View File

@ -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)

View File

@ -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:

View File

@ -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:

View File

@ -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:
" <Function 'test_func'>",
])
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):

View File

@ -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))