[py-svn] r33387 - in py/dist/py/test/rsession: . testing
fijal at codespeak.net
fijal at codespeak.net
Tue Oct 17 20:19:16 CEST 2006
Author: fijal
Date: Tue Oct 17 20:18:52 2006
New Revision: 33387
Modified:
py/dist/py/test/rsession/executor.py
py/dist/py/test/rsession/local.py
py/dist/py/test/rsession/report.py
py/dist/py/test/rsession/rsession.py
py/dist/py/test/rsession/testing/test_lsession.py
Log:
Several fixes, extending Lsession.
Modified: py/dist/py/test/rsession/executor.py
==============================================================================
--- py/dist/py/test/rsession/executor.py (original)
+++ py/dist/py/test/rsession/executor.py Tue Oct 17 20:18:52 2006
@@ -3,16 +3,19 @@
import py, os
-from py.__.test.rsession.outcome import Outcome
+from py.__.test.rsession.outcome import Outcome, ReprOutcome
from py.__.test.rsession.box import Box
+from py.__.test.rsession import report
class RunExecutor(object):
""" Same as in executor, but just running run
"""
wraps = False
- def __init__(self, item):
+ def __init__(self, item, usepdb=False, reporter=None):
self.item = item
+ self.usepdb = usepdb
+ self.reporter = reporter
def execute(self):
try:
@@ -20,6 +23,8 @@
outcome = Outcome()
except py.test.Item.Skipped, e:
outcome = Outcome(skipped=str(e))
+ except (KeyboardInterrupt, SystemExit):
+ raise
except:
excinfo = py.code.ExceptionInfo()
if isinstance(self.item, py.test.Function):
@@ -28,6 +33,15 @@
excinfo.traceback = excinfo.traceback.cut(
path=code.path, firstlineno=code.firstlineno)
outcome = Outcome(excinfo=excinfo, setupfailure=False)
+ if self.usepdb:
+ if self.reporter is not None:
+ self.reporter(report.ImmediateFailure(self.item,
+ ReprOutcome(outcome.make_repr())))
+ import pdb
+ pdb.post_mortem(excinfo._excinfo[2])
+ # XXX hmm, we probably will not like to continue from that point
+ # or we do?
+ raise SystemExit()
outcome.stdout = ""
outcome.stderr = ""
return outcome
@@ -75,4 +89,3 @@
if excinfo is not None:
return Outcome(excinfo=excinfo, setupfailure=False)
return Outcome()
-
Modified: py/dist/py/test/rsession/local.py
==============================================================================
--- py/dist/py/test/rsession/local.py (original)
+++ py/dist/py/test/rsession/local.py Tue Oct 17 20:18:52 2006
@@ -2,22 +2,43 @@
""" local-only operations
"""
-from py.__.test.rsession.executor import BoxExecutor
+from py.__.test.rsession.executor import BoxExecutor, RunExecutor
from py.__.test.rsession import report
from py.__.test.rsession.outcome import ReprOutcome
-def run(item):
+def box_runner(item, config, reporter):
r = BoxExecutor(item)
return ReprOutcome(r.execute())
-def local_loop(reporter, itemgenerator, shouldstop):
-
+def plain_runner(item, config, reporter):
+ r = RunExecutor(item, usepdb=config.option.usepdb, reporter=reporter)
+ return ReprOutcome(r.execute().make_repr())
+
+def benchmark_runner(item, config, reporter):
+ raise NotImplementedError()
+
+def apigen_runner(item, config, reporter):
+ pass
+
+def exec_runner(item, config, reporter):
+ raise NotImplementedError()
+
+# runner interface is here to perform several different types of run
+#+1. box_runner - for running normal boxed tests
+#+2. plain_runner - for performing running without boxing (necessary for pdb)
+# XXX: really?
+#-3. exec_runner - for running under different interpreter
+#-4. benchmark_runner - for running with benchmarking
+#-5. apigen_runner - for running under apigen to generate api out of it.
+def local_loop(reporter, itemgenerator, shouldstop, config, runner=None):
+ if runner is None:
+ runner = box_runner
while 1:
try:
item = itemgenerator.next()
if shouldstop():
return
- outcome = run(item)
+ outcome = runner(item, config, reporter)
reporter(report.ReceivedItemOutcome(None, item, outcome))
except StopIteration:
break
Modified: py/dist/py/test/rsession/report.py
==============================================================================
--- py/dist/py/test/rsession/report.py (original)
+++ py/dist/py/test/rsession/report.py Tue Oct 17 20:18:52 2006
@@ -111,3 +111,8 @@
class RsyncFinished(ReportEvent):
def __init__(self):
self.time = time.time()
+
+class ImmediateFailure(ReportEvent):
+ def __init__(self, item, outcome):
+ self.item = item
+ self.outcome = outcome
Modified: py/dist/py/test/rsession/rsession.py
==============================================================================
--- py/dist/py/test/rsession/rsession.py (original)
+++ py/dist/py/test/rsession/rsession.py Tue Oct 17 20:18:52 2006
@@ -16,7 +16,7 @@
from py.__.test.rsession.hostmanage import init_hosts, teardown_hosts
from py.__.test.terminal.out import getout
-from py.__.test.rsession.local import local_loop
+from py.__.test.rsession.local import local_loop, plain_runner
class RemoteOptions(object):
def __init__(self, d):
@@ -77,7 +77,11 @@
def report_RsyncFinished(self, item):
self.timersync = item.time
+ def report_ImmediateFailure(self, event):
+ self.repr_failure(event.item, event.outcome)
+
def report_TestFinished(self, item):
+ self.out.line()
assert hasattr(self, 'timestart')
self.timeend = item.timeend
self.skips()
@@ -109,8 +113,8 @@
def repr_failure(self, item, outcome):
excinfo = outcome.excinfo
traceback = excinfo.traceback
- if item and not self.config.option.fulltrace:
- path, firstlineno = item.getpathlineno()
+ #if item and not self.config.option.fulltrace:
+ # path, firstlineno = item.getpathlineno()
if not traceback:
self.out.line("empty traceback from item %r" % (item,))
return
@@ -395,7 +399,7 @@
class LSession(AbstractSession):
""" Local version of session
"""
- def main(self, args, reporter=None):
+ def main(self, args, reporter=None, runner=None):
if not args:
args = [py.path.local()]
@@ -419,7 +423,10 @@
itemgenerator = itemgen()
#assert 0, "\n".join([",".join(x.listnames()) for x in
# list(itemgenerator)])
- local_loop(reporter, itemgenerator, checkfun)
+ # XXX: We have to decide which runner to use at this point
+ if runner is None and self.config.option.usepdb:
+ runner = plain_runner
+ local_loop(reporter, itemgenerator, checkfun, self.config, runner=runner)
reporter(report.TestFinished())
if startserverflag:
Modified: py/dist/py/test/rsession/testing/test_lsession.py
==============================================================================
--- py/dist/py/test/rsession/testing/test_lsession.py (original)
+++ py/dist/py/test/rsession/testing/test_lsession.py Tue Oct 17 20:18:52 2006
@@ -5,9 +5,10 @@
import py
from py.__.test.rsession.rsession import LSession
from py.__.test.rsession import report
+from py.__.test.rsession.local import box_runner, plain_runner
class TestLSession(object):
- def test_example_distribution(self):
+ def example_distribution(self, runner):
# XXX find a better way for the below
tmpdir = py.path.local(py.__file__).dirpath().dirpath()
tmpdir.ensure("sub", "__init__.py")
@@ -20,20 +21,26 @@
raise ValueError(23)
def test_4(someargs):
pass
+ #def test_5():
+ # import os
+ # os.kill(os.getpid(), 11)
"""))
args = [str(tmpdir.join("sub"))]
config, args = py.test.Config.parse(args)
lsession = LSession(config)
allevents = []
- lsession.main(args, reporter=allevents.append)
+ lsession.main(args, reporter=allevents.append, runner=runner)
testevents = [x for x in allevents
if isinstance(x, report.ReceivedItemOutcome)]
assert len(testevents)
passevents = [i for i in testevents if i.outcome.passed]
failevents = [i for i in testevents if i.outcome.excinfo]
skippedevents = [i for i in testevents if i.outcome.skipped]
+ signalevents = [i for i in testevents if i.outcome.signal]
assert len(passevents) == 1
assert len(failevents) == 3
+ assert len(skippedevents) == 0
+ #assert len(signalevents) == 1
tb = failevents[0].outcome.excinfo.traceback
assert str(tb[0].path).find("test_one") != -1
assert str(tb[0].source).find("test_2") != -1
@@ -47,3 +54,9 @@
assert failevents[2].outcome.excinfo.typename == 'TypeError'
assert str(tb[0].path).find("executor") != -1
assert str(tb[0].source).find("execute") != -1
+
+ def test_normal(self):
+ self.example_distribution(box_runner)
+
+ def test_plain(self):
+ self.example_distribution(plain_runner)
More information about the pytest-commit
mailing list