[py-svn] r56685 - in py/branch/event/py/test2: . rsession testing
hpk at codespeak.net
hpk at codespeak.net
Sun Jul 20 16:18:20 CEST 2008
Author: hpk
Date: Sun Jul 20 16:18:18 2008
New Revision: 56685
Added:
py/branch/event/py/test2/runner.py
- copied, changed from r56684, py/branch/event/py/test2/executor.py
py/branch/event/py/test2/testing/test_runner.py
- copied, changed from r56684, py/branch/event/py/test2/testing/test_executor.py
Removed:
py/branch/event/py/test2/executor.py
py/branch/event/py/test2/testing/test_executor.py
Modified:
py/branch/event/py/test2/item.py
py/branch/event/py/test2/rsession/slave.py
py/branch/event/py/test2/session.py
Log:
executor classes -> simpler runner functions -> 30% less code in that module :)
Deleted: /py/branch/event/py/test2/executor.py
==============================================================================
--- /py/branch/event/py/test2/executor.py Sun Jul 20 16:18:18 2008
+++ (empty file)
@@ -1,92 +0,0 @@
-"""
- test run functions
-"""
-
-import py, os, sys
-
-from py.__.test2 import repevent
-from py.__.test2.outcome import Skipped
-from py.__.test.outcome import Skipped as Skipped2
-import py.__.test2.custompdb
-
-class RunExecutor(object):
- """
- executes a test item and fills in a ItemTestReport.
- """
- def __init__(self, item):
- self.item = item
- self.config = item._config
- self.outerr = None
-
- def runitem(self):
- capture = self.config._getcapture(path=self.item.fspath)
- try:
- self.item.run()
- finally:
- out, err = capture.reset()
- self.outerr = (("recorded stdout", out), ("recorded stderr", err))
-
- def execute(self):
- try:
- self.runitem()
- except py.builtin.sysex:
- raise
- except:
- excinfo = py.code.ExceptionInfo()
- if self.config.option.usepdb and not excinfo.errisinstance(Skipped):
- py.__.test2.custompdb.post_mortem(excinfo._excinfo[2])
- else:
- excinfo = None
- return self._makereport(excinfo)
-
- def _makereport(self, excinfo):
- trail = self.item._get_collector_trail()
- repr_run = None
- if excinfo is not None:
- if excinfo.errisinstance((Skipped, Skipped2)):
- outcome = "skipped"
- else:
- outcome = "failed"
- else:
- outcome = "passed"
- runinfo = RunInfo(excinfo, self.outerr)
- repr_run = self.item.repr_run(runinfo)
- repr_path = self.item.repr_path()
- return repevent.ItemTestReport(trail, outcome, repr_run, repr_path)
-
-class RunInfo:
- def __init__(self, excinfo, outerr):
- self.excinfo = excinfo
- self.outerr = outerr
-
-class ForkExecutor(RunExecutor):
- """ Same as RunExecutor, but boxes test instead
- """
- def execute_forked(self):
- testrep = RunExecutor.execute(self)
- return testrep.dumps()
-
- def execute(self):
- ff = py.io.ForkedFunc(self.execute_forked)
- result = ff.waitfinish()
- return self.maketestreport(result)
-
- def maketestreport(self, result):
- if result.retval is not None:
- testrep = repevent.ItemTestReport.fromdumps(result.retval)
- testrep.stdout = result.out
- testrep.stderr = result.err
- else:
- testrep = repevent.ItemTestReport.fromitem(self.item, None, None)
- testrep.outcome = "failed"
- testrep.stdout = result.out
- testrep.stderr = result.err
- testrep.signal = result.err
- return testrep
-
-def getexecutor(item):
- cls = RunExecutor
- if item._config.option.boxed:
- cls = ForkExecutor
- return cls(item)
-
Modified: py/branch/event/py/test2/item.py
==============================================================================
--- py/branch/event/py/test2/item.py (original)
+++ py/branch/event/py/test2/item.py Sun Jul 20 16:18:18 2008
@@ -1,5 +1,6 @@
import py
from py.__.test2.collect import FunctionMixin, Node
+from py.__.test2.runner import basic_runner, fork_runner
from py.__.test2 import present
_dummy = object()
@@ -44,13 +45,10 @@
"""
xxx
- def runtest(self):
- """ Return an ItemTestReport event which has all the
- information about running the underlying test item.
- """
- from py.__.test2.executor import getexecutor
- executor = getexecutor(self)
- return executor.execute()
+ def _getrunner(self):
+ if self._config.option.boxed:
+ return fork_runner
+ return basic_runner
class Function(FunctionMixin, Item):
""" a Function Item is responsible for setting up
Modified: py/branch/event/py/test2/rsession/slave.py
==============================================================================
--- py/branch/event/py/test2/rsession/slave.py (original)
+++ py/branch/event/py/test2/rsession/slave.py Sun Jul 20 16:18:18 2008
@@ -13,7 +13,8 @@
send(None)
break
item = config._getcollector(itemspec)
- testrep = item.runtest()
+ runner = item._getrunner()
+ testrep = runner(item)
send(testrep.dumps())
def setup():
Copied: py/branch/event/py/test2/runner.py (from r56684, py/branch/event/py/test2/executor.py)
==============================================================================
--- py/branch/event/py/test2/executor.py (original)
+++ py/branch/event/py/test2/runner.py Sun Jul 20 16:18:18 2008
@@ -9,84 +9,62 @@
from py.__.test.outcome import Skipped as Skipped2
import py.__.test2.custompdb
-class RunExecutor(object):
- """
- executes a test item and fills in a ItemTestReport.
- """
- def __init__(self, item):
- self.item = item
- self.config = item._config
- self.outerr = None
+class RunInfo:
+ """ info on test runs. """
+ def __init__(self, excinfo, outerr):
+ self.excinfo = excinfo
+ self.outerr = outerr
- def runitem(self):
- capture = self.config._getcapture(path=self.item.fspath)
+def basic_runner(item):
+ """ returns a test report after having run the given test item. """
+ config = item._config
+ capture = config._getcapture(path=item.fspath)
+ excinfo = None
+ try:
try:
- self.item.run()
+ item.run()
finally:
out, err = capture.reset()
- self.outerr = (("recorded stdout", out), ("recorded stderr", err))
-
- def execute(self):
- try:
- self.runitem()
- except py.builtin.sysex:
- raise
- except:
- excinfo = py.code.ExceptionInfo()
- if self.config.option.usepdb and not excinfo.errisinstance(Skipped):
- py.__.test2.custompdb.post_mortem(excinfo._excinfo[2])
+ outerr = (("recorded stdout", out), ("recorded stderr", err))
+ except py.builtin.sysex:
+ raise
+ except:
+ excinfo = py.code.ExceptionInfo()
+ if config.option.usepdb and not excinfo.errisinstance(Skipped):
+ py.__.test2.custompdb.post_mortem(excinfo._excinfo[2])
+
+ # make report
+ trail = item._get_collector_trail()
+ repr_run = None
+ if excinfo is not None:
+ if excinfo.errisinstance((Skipped, Skipped2)):
+ outcome = "skipped"
else:
- excinfo = None
- return self._makereport(excinfo)
+ outcome = "failed"
+ else:
+ outcome = "passed"
+ runinfo = RunInfo(excinfo, outerr)
+ repr_run = item.repr_run(runinfo)
+ repr_path = item.repr_path()
+ return repevent.ItemTestReport(trail, outcome, repr_run, repr_path)
- def _makereport(self, excinfo):
- trail = self.item._get_collector_trail()
- repr_run = None
- if excinfo is not None:
- if excinfo.errisinstance((Skipped, Skipped2)):
- outcome = "skipped"
- else:
- outcome = "failed"
- else:
- outcome = "passed"
- runinfo = RunInfo(excinfo, self.outerr)
- repr_run = self.item.repr_run(runinfo)
- repr_path = self.item.repr_path()
- return repevent.ItemTestReport(trail, outcome, repr_run, repr_path)
-class RunInfo:
- def __init__(self, excinfo, outerr):
- self.excinfo = excinfo
- self.outerr = outerr
-
-class ForkExecutor(RunExecutor):
- """ Same as RunExecutor, but boxes test instead
- """
- def execute_forked(self):
- testrep = RunExecutor.execute(self)
+def fork_runner(item):
+ def runforked():
+ testrep = basic_runner(item)
return testrep.dumps()
- def execute(self):
- ff = py.io.ForkedFunc(self.execute_forked)
- result = ff.waitfinish()
- return self.maketestreport(result)
-
- def maketestreport(self, result):
- if result.retval is not None:
- testrep = repevent.ItemTestReport.fromdumps(result.retval)
- testrep.stdout = result.out
- testrep.stderr = result.err
- else:
- testrep = repevent.ItemTestReport.fromitem(self.item, None, None)
- testrep.outcome = "failed"
- testrep.stdout = result.out
- testrep.stderr = result.err
- testrep.signal = result.err
- return testrep
-
-def getexecutor(item):
- cls = RunExecutor
- if item._config.option.boxed:
- cls = ForkExecutor
- return cls(item)
+ ff = py.io.ForkedFunc(runforked)
+ result = ff.waitfinish()
+ if result.retval is not None:
+ testrep = repevent.ItemTestReport.fromdumps(result.retval)
+ testrep.stdout = result.out
+ testrep.stderr = result.err
+ else:
+ testrep = repevent.ItemTestReport.fromitem(self.item, None, None)
+ testrep.outcome = "failed"
+ testrep.stdout = result.out
+ testrep.stderr = result.err
+ testrep.signal = result.err
+ return testrep
Modified: py/branch/event/py/test2/session.py
==============================================================================
--- py/branch/event/py/test2/session.py (original)
+++ py/branch/event/py/test2/session.py Sun Jul 20 16:18:18 2008
@@ -84,5 +84,6 @@
return failures
def runtest(self, item):
- return item.runtest()
+ runner = item._getrunner()
+ return runner(item)
Deleted: /py/branch/event/py/test2/testing/test_executor.py
==============================================================================
--- /py/branch/event/py/test2/testing/test_executor.py Sun Jul 20 16:18:18 2008
+++ (empty file)
@@ -1,168 +0,0 @@
-
-import py
-from py.__.test2.executor import RunExecutor, ForkExecutor
-
-def setup_module(mod):
- mod.tmpdir = py.test.ensuretemp(mod.__name__)
-
-class TestExecutor:
- Executor = RunExecutor
-
- def setup_method(self, method):
- self.tmpdir = tmpdir.join("%s_%s" %
- (self.__class__.__name__, method.__name__))
-
- def makeitem(self, func, funcname="testfunc"):
- funcname = getattr(func, '__name__', funcname)
- self.tmpdir.ensure("__init__.py")
- path = self.tmpdir.ensure(funcname + ".py")
- path.write(py.code.Source(func))
- self.config = py.test2.config._reparse([path.dirpath()])
- modulecol = self.config._getcollector(path)
- item = modulecol.join(funcname)
- assert item is not None, (item, funcname)
- return item
-
- def exectestfunc(self, func, funcname="testfunc"):
- item = self.makeitem(func, funcname=funcname)
- executor = self.getexec(item)
- return executor.execute()
-
- def getexec(self, item):
- return self.Executor(item)
-
- def test_run_executor_pass(self):
- testrep = self.exectestfunc("""
- def testfunc():
- pass
- """)
- assert testrep.passed
-
- def test_run_executor_fail(self):
- testrep = self.exectestfunc("""
- def testfunc():
- assert 0
- """)
- assert not testrep.passed
- assert testrep.failed
-
- def test_run_executor_skip(self):
- testrep = self.exectestfunc("""
- import py
- def testfunc():
- py.test2.skip("hello")
- """)
- assert testrep.skipped
-
- def test_run_executor_capture_stdout(self):
- testrep = self.exectestfunc("""
- def testfunc():
- print "samfing"
- """)
- assert testrep.passed
- assert testrep.repr_run.find("stderr") == -1
- assert testrep.repr_run.find("samfing\n") != -1
-
- def test_run_executor_capture_stderr(self):
- testrep = self.exectestfunc("""
- import sys
- def testfunc():
- print >>sys.stderr, "samfong"
- """)
- print testrep.repr_run
- assert testrep.repr_run.find("stdout") == -1
- assert testrep.repr_run.find("stderr") != -1
- assert testrep.repr_run.find("samfong\n") != -1
-
- def test_executor_print_with_assertion_failed(self):
- testrep = self.exectestfunc("""
- import sys
- def testfunc():
- print "samfing elz"
- print >>sys.stderr, "sure"
- assert 0
- """)
- assert not testrep.passed
- assert testrep.failed
- assert testrep.repr_run.find("stdout") != -1
- assert testrep.repr_run.find("stderr") != -1
- assert testrep.repr_run.find("samfing elz") != -1
- assert testrep.repr_run.find("sure") != -1
-
- def test_executor_print_with_explicit_fail(self):
- testrep = self.exectestfunc("""
- import py, sys
- def testfunc():
- print "samfing elz"
- print >>sys.stderr, "sure"
- py.test.fail()
- """)
- assert not testrep.passed
- assert testrep.failed
- assert testrep.repr_run.find("stdout") != -1
- assert testrep.repr_run.find("stderr") != -1
- assert testrep.repr_run.find("samfing elz\n") != -1
- assert testrep.repr_run.find("sure\n") != -1
-
- def test_executor_setupfailure(self):
- py.test.skip("implement setupfailures")
- testrep = self.exectestfunc("""
- def setup_module(mod):
- raise ValueError(10)
- def testfunc():
- pass
- """)
- assert not testrep.passed
- assert testrep.setupfailure
- assert testrep.failed # ???
-
- # xxx repr_run of setupfailures should always
- # display python Tracebacks starting from the
- # failing setup function
-
- def test_executor_setupfailure_on_eager_teardown(self):
- py.test.skip("implement setupfailures")
- testrep = self.exectestfunc("""
- def testfunc():
- pass
- def teardown_function(func):
- raise ValueError(11)
- """)
- assert not testrep.passed
- assert testrep.setupfailure
- assert testrep.failed # ???
- # xxx repr_run of setupfailures should always
- # display python Tracebacks starting from the
- # failing teardown function
-
- def test_executor_setupfailure_on_deferred_teardown(self):
- py.test.skip("implement setupfailures, extend help machinery")
- testrep = self.exectestfunc("""
- class TestA:
- def testfunc(self):
- pass
- def teardown_class(cls):
- raise ValueError(12)
- def setup_function(self, method):
- raise ValueError(11)
- def testfunc():
- pass
- """, "TestA.testfunc", "testfunc")
-
-class TestForkExecutor(TestExecutor):
- def setup_class(cls):
- if not hasattr(py.std.os, 'fork'):
- py.test.skip("need os.fork()")
- Executor = ForkExecutor
-
- def test_suicide(self):
- py.test.skip("XXX")
- testrep = self.exectestfunc("""
- import os
- def testfunc():
- os.kill(15, os.getpid())
- """)
- assert testrep.signal == 15
- assert not testrep.passed
- assert testrep.failed
-
Copied: py/branch/event/py/test2/testing/test_runner.py (from r56684, py/branch/event/py/test2/testing/test_executor.py)
==============================================================================
--- py/branch/event/py/test2/testing/test_executor.py (original)
+++ py/branch/event/py/test2/testing/test_runner.py Sun Jul 20 16:18:18 2008
@@ -1,12 +1,13 @@
import py
-from py.__.test2.executor import RunExecutor, ForkExecutor
+from py.__.test2.runner import basic_runner, fork_runner
def setup_module(mod):
mod.tmpdir = py.test.ensuretemp(mod.__name__)
-class TestExecutor:
- Executor = RunExecutor
+class TestRunner:
+ def getrunner(self):
+ return basic_runner
def setup_method(self, method):
self.tmpdir = tmpdir.join("%s_%s" %
@@ -23,39 +24,37 @@
assert item is not None, (item, funcname)
return item
- def exectestfunc(self, func, funcname="testfunc"):
+ def runtestfunc(self, func, funcname="testfunc"):
item = self.makeitem(func, funcname=funcname)
- executor = self.getexec(item)
- return executor.execute()
+ runner = self.getrunner()
+ return runner(item)
- def getexec(self, item):
- return self.Executor(item)
- def test_run_executor_pass(self):
- testrep = self.exectestfunc("""
+ def test_run_pass(self):
+ testrep = self.runtestfunc("""
def testfunc():
pass
""")
assert testrep.passed
- def test_run_executor_fail(self):
- testrep = self.exectestfunc("""
+ def test_run_fail(self):
+ testrep = self.runtestfunc("""
def testfunc():
assert 0
""")
assert not testrep.passed
assert testrep.failed
- def test_run_executor_skip(self):
- testrep = self.exectestfunc("""
+ def test_run_skip(self):
+ testrep = self.runtestfunc("""
import py
def testfunc():
py.test2.skip("hello")
""")
assert testrep.skipped
- def test_run_executor_capture_stdout(self):
- testrep = self.exectestfunc("""
+ def test_run_capture_stdout(self):
+ testrep = self.runtestfunc("""
def testfunc():
print "samfing"
""")
@@ -63,8 +62,8 @@
assert testrep.repr_run.find("stderr") == -1
assert testrep.repr_run.find("samfing\n") != -1
- def test_run_executor_capture_stderr(self):
- testrep = self.exectestfunc("""
+ def test_run_capture_stderr(self):
+ testrep = self.runtestfunc("""
import sys
def testfunc():
print >>sys.stderr, "samfong"
@@ -74,8 +73,8 @@
assert testrep.repr_run.find("stderr") != -1
assert testrep.repr_run.find("samfong\n") != -1
- def test_executor_print_with_assertion_failed(self):
- testrep = self.exectestfunc("""
+ def test_print_with_assertion_failed(self):
+ testrep = self.runtestfunc("""
import sys
def testfunc():
print "samfing elz"
@@ -89,8 +88,8 @@
assert testrep.repr_run.find("samfing elz") != -1
assert testrep.repr_run.find("sure") != -1
- def test_executor_print_with_explicit_fail(self):
- testrep = self.exectestfunc("""
+ def test_print_with_explicit_fail(self):
+ testrep = self.runtestfunc("""
import py, sys
def testfunc():
print "samfing elz"
@@ -104,9 +103,9 @@
assert testrep.repr_run.find("samfing elz\n") != -1
assert testrep.repr_run.find("sure\n") != -1
- def test_executor_setupfailure(self):
+ def test_setupfailure(self):
py.test.skip("implement setupfailures")
- testrep = self.exectestfunc("""
+ testrep = self.runtestfunc("""
def setup_module(mod):
raise ValueError(10)
def testfunc():
@@ -120,9 +119,9 @@
# display python Tracebacks starting from the
# failing setup function
- def test_executor_setupfailure_on_eager_teardown(self):
+ def test_setupfailure_on_eager_teardown(self):
py.test.skip("implement setupfailures")
- testrep = self.exectestfunc("""
+ testrep = self.runtestfunc("""
def testfunc():
pass
def teardown_function(func):
@@ -135,9 +134,9 @@
# display python Tracebacks starting from the
# failing teardown function
- def test_executor_setupfailure_on_deferred_teardown(self):
+ def test_setupfailure_on_deferred_teardown(self):
py.test.skip("implement setupfailures, extend help machinery")
- testrep = self.exectestfunc("""
+ testrep = self.runtestfunc("""
class TestA:
def testfunc(self):
pass
@@ -149,15 +148,16 @@
pass
""", "TestA.testfunc", "testfunc")
-class TestForkExecutor(TestExecutor):
+class TestForkRunner(TestRunner):
def setup_class(cls):
if not hasattr(py.std.os, 'fork'):
py.test.skip("need os.fork()")
- Executor = ForkExecutor
+ def getrunner(self):
+ return fork_runner
def test_suicide(self):
py.test.skip("XXX")
- testrep = self.exectestfunc("""
+ testrep = self.runtestfunc("""
import os
def testfunc():
os.kill(15, os.getpid())
More information about the pytest-commit
mailing list