[py-svn] r7113 - in py/dist/py: . execnet path/svn test test/report test/report/text
hpk at codespeak.net
hpk at codespeak.net
Sun Oct 24 16:42:30 CEST 2004
Author: hpk
Date: Sun Oct 24 16:42:27 2004
New Revision: 7113
Added:
py/dist/py/execnet/channel.py
py/dist/py/test/item.py
Modified:
py/dist/py/__init__.py
py/dist/py/execnet/gateway.py
py/dist/py/path/svn/svntestbase.py
py/dist/py/path/svn/test_urlcommand.py
py/dist/py/path/svn/test_wccommand.py
py/dist/py/test/cmdline.py
py/dist/py/test/collect.py
py/dist/py/test/compat.py
py/dist/py/test/raises.py
py/dist/py/test/report/memo.py
py/dist/py/test/report/text/reporter.py
py/dist/py/test/report/text/summary.py
py/dist/py/test/run.py
py/dist/py/test/test_compat.py
Log:
pretty large refactoring
most of the testing process is now outsourced
into a child process. This is the foundation
to implement a "test session" which will rerun
the set of failing tests. Note that without
running the tests in a child process it's somewhat
hard to restart the testing process because
we can't easily "reload()" changed code.
also there are some API enhancements (which were
purposedly not described in the documentation anyway)
most notably, there now are direct
py.test.fail("my message")
py.test.skip("my reason")
py.test.exit("why i want it immediately")
functions, so you don't have to raise some
exception anymore.
also the exposed namespace from py.test is
somewhat smaller and cleaner now.
there are still some problems with the
shutdown procedure but nothing too serious.
Modified: py/dist/py/__init__.py
==============================================================================
--- py/dist/py/__init__.py (original)
+++ py/dist/py/__init__.py Sun Oct 24 16:42:27 2004
@@ -16,17 +16,17 @@
'test.collect.Module': './test/collect.Module',
'test.collect.PyCollector':'./test/collect.PyCollector',
'test.collect.Error': './test/collect.Error',
- 'test.Failed': './test/run.Failed',
- 'test.Passed': './test/run.Passed',
- 'test.Skipped': './test/run.Skipped',
- 'test.raises': './test/raises.raises',
- 'test.config': './test/config.config',
- 'test.compat.TestCase': './test/compat.TestCase',
+ 'test.Item': './test/item.Item',
'test.Driver': './test/run.Driver',
- 'test.Item': './test/run.Item',
'test.Option': './test/tool/optparse.Option',
'test.TextReporter': './test/report/text/reporter.TextReporter',
'test.MemoReporter': './test/report/memo.MemoReporter',
+ 'test.exit': './test/run.exit',
+ 'test.fail': './test/run.fail',
+ 'test.skip': './test/run.skip',
+ 'test.raises': './test/raises.raises',
+ 'test.config': './test/config.config',
+ 'test.compat.TestCase': './test/compat.TestCase',
'process.cmdexec': './process/cmdexec.cmdexec',
Added: py/dist/py/execnet/channel.py
==============================================================================
--- (empty file)
+++ py/dist/py/execnet/channel.py Sun Oct 24 16:42:27 2004
@@ -0,0 +1,31 @@
+
+class ChannelFile:
+ def __init__(self, channel):
+ self.channel = channel
+
+ def write(self, out):
+ if self.channel.isclosed():
+ raise IOError, "cannot write to %r" % self
+ self.channel.send(out)
+
+ def flush(self):
+ pass
+
+ def close(self):
+ if self.channel.isclosed():
+ raise IOError, "cannot close %r" % self
+ self.channel.close()
+
+ def __repr__(self):
+ state = self.channel.isclosed() and 'closed' or 'open'
+ return '<ChannelFile %d %s>' %(self.channel.id, state)
+
+def receive2file(channel, f):
+ while 1:
+ try:
+ out = channel.receive()
+ except EOFError:
+ break
+ f.write(out)
+ f.flush()
+
Modified: py/dist/py/execnet/gateway.py
==============================================================================
--- py/dist/py/execnet/gateway.py (original)
+++ py/dist/py/execnet/gateway.py Sun Oct 24 16:42:27 2004
@@ -8,13 +8,11 @@
debug = 0
sysex = (KeyboardInterrupt, SystemExit)
-class Closed:
- pass
-
-class RemoteError(Closed):
+class RemoteError(EOFError):
""" Contains an Exceptions from the other side. """
def __init__(self, formatted):
self.formatted = formatted
+ EOFError.__init__(self)
def __str__(self):
return self.formatted
@@ -175,9 +173,12 @@
self._depchannel = []
def __repr__(self):
- flag = self._closeevent.isSet() and "closed" or "open"
+ flag = self.isclosed() and "closed" or "open"
return "<Channel id=%d %s>" % (self.id, flag)
+ def isclosed(self):
+ return self._closeevent.isSet()
+
def close(self, error=None):
""" close down this channel on both sides. """
put = self.gateway._outgoing.put
@@ -187,10 +188,10 @@
put(Message.CHANNEL_CLOSE(self.id))
self._close()
- def _close(self, finalitem=Closed()):
+ def _close(self, finalitem=EOFError()):
self._finalitem = finalitem
for x in self._depchannel:
- x._close(Closed())
+ x._close()
del self.gateway.channelfactory[self.id]
self._items.put(finalitem)
self._closeevent.set()
@@ -242,7 +243,7 @@
a textual representation of the remote traceback.
"""
x = self._items.get()
- if isinstance(x, Closed):
+ if isinstance(x, EOFError):
raise x
return x
Modified: py/dist/py/path/svn/svntestbase.py
==============================================================================
--- py/dist/py/path/svn/svntestbase.py (original)
+++ py/dist/py/path/svn/svntestbase.py Sun Oct 24 16:42:27 2004
@@ -9,8 +9,8 @@
bn = meth.func_name
for x in 'test_remove', 'test_move':
if bn.startswith(x):
- raise py.test.Skipped(msg=
- "tests for modifying svn needs better test state management")
+ py.test.skip(
+ "tests for modifying svn needs better test state management")
def test_propget(self):
url = self.root.join("samplefile")
Modified: py/dist/py/path/svn/test_urlcommand.py
==============================================================================
--- py/dist/py/path/svn/test_urlcommand.py (original)
+++ py/dist/py/path/svn/test_urlcommand.py Sun Oct 24 16:42:27 2004
@@ -12,7 +12,7 @@
raise py.test.Skipped(msg="XXX fix svnurl first")
def xtest_copy_dir(self):
- raise py.test.Skipped(msg="XXX fix svnurl first")
+ py.test.skipp("XXX fix svnurl first")
def XXXtest_info_log(self):
url = self.root.join("samplefile")
Modified: py/dist/py/path/svn/test_wccommand.py
==============================================================================
--- py/dist/py/path/svn/test_wccommand.py (original)
+++ py/dist/py/path/svn/test_wccommand.py Sun Oct 24 16:42:27 2004
@@ -15,7 +15,7 @@
py.process.cmdexec('svnadmin create %s' % repo)
except py.process.cmdexec.Error:
repo.remove()
- raise py.test.Skipped(msg='could not create temporary svn test repository')
+ raise py.test.skip('could not create temporary svn test repository')
wcdir.ensure(dir=1)
print "created svn repository", repo
wc = py.path.svnwc(wcdir)
Modified: py/dist/py/test/cmdline.py
==============================================================================
--- py/dist/py/test/cmdline.py (original)
+++ py/dist/py/test/cmdline.py Sun Oct 24 16:42:27 2004
@@ -1,5 +1,7 @@
from __future__ import generators
import py
+import sys
+from py.__impl__.execnet.channel import ChannelFile, receive2file
#
# main entry point
@@ -16,39 +18,101 @@
import __main__
return [py.test.collect.Module(py.std.sys.argv[0])]
+def waitfilechange():
+ """ wait until project files are changed. """
+ pkgdir = py.test.config.getfirst('pkgdir')
+ pkgdir = py.path.local(pkgdir)
+ fil = py.path.checker(fnmatch='*.py')
+ rec = py.path.checker(dotfile=0)
+ statcache = {}
+ for path in pkgdir.visit(fil, rec):
+ statcache[path] = path.stat()
+
+ while 1:
+ py.std.time.sleep(0.2)
+ for path in pkgdir.visit(fil, rec):
+ st = path.stat()
+ cst = statcache[path]
+ if st.st_mtime != cst.st_mtime or \
+ st.st_size != cst.st_size:
+ return
+
+class FailingCollector(py.test.collect.Collector):
+ def __init__(self, faileditems):
+ self._faileditems = faileditems
+
+ def __iter__(self):
+ for x in self._faileditems:
+ yield x
+
+def master():
+ #if not py.test.config.option.session:
+ # break
+ #l = driver.getfailed()
+ #if l:
+ # collectors = [FailingCollector(l)]
+ #elif isinstance(fncollectors[0], FailingCollector):
+ # collectors = fncollectors
+ #else:
+ # break
+ waitfilechange()
+
def main():
args = py.std.sys.argv[1:]
py.test.config.readconfiguration(*getanchors(args))
- if py.test.config.restartpython():
- return
filenames = py.test.config.parseargs(args)
- collectors = getcollectors(filenames)
+ if not filenames:
+ filenames.append(str(py.path.local()))
- reporter = py.test.config.getfirst('getreporter') ()
- driver = py.test.Driver(reporter)
- driver.setup()
- try:
+ gw = py.execnet.PopenGateway()
+ channel = gw.remote_exec("""
+ from py.__impl__.test.cmdline import slave
+ slave(channel)
+ """)
+ print "sending (args, filenames)"
+ channel.send((args, filenames))
+ print "receiving stdout/stderr"
+ stdout = channel.receive()
+ stderr = channel.receive()
+ py.std.threading.Thread(target=receive2file,
+ args=(stdout, sys.stdout)).start()
+ py.std.threading.Thread(target=receive2file,
+ args=(stderr, sys.stderr)).start()
+ print "waiting"
+ while 1:
try:
- reporter.start()
- try:
- for collector in collectors:
- driver.run_collector_or_item(collector)
- except driver.Exit:
- pass
- finally:
- driver.teardown()
- except KeyboardInterrupt:
- print >>py.std.sys.stderr, "KEYBOARD INTERRUPT"
- py.std.sys.exit(2)
- except SystemExit:
- print >>py.std.sys.stderr, "SYSTEM Exit"
- py.std.sys.exit(-1)
- reporter.end()
-
+ channel.waitclose(0.1)
+ except IOError:
+ continue
+ except KeyboardInterrupt:
+ print
+ print "keyboard interrupt"
+ except:
+ py.std.traceback.print_exc()
+ channel.close()
+ gw.exit()
+ break
+
+def slave(channel):
+ """ we run this on the other side. """
+ args, filenames = channel.receive()
+ out, err = channel.newchannel(), channel.newchannel()
+ channel.send(out)
+ channel.send(err)
+ sys.stdout = ChannelFile(out)
+ sys.stderr = ChannelFile(err)
+
+ filenames = map(py.path.local, filenames)
+ py.test.config.readconfiguration(*filenames)
+ py.test.config.parseargs(args)
+ fncollectors = list(getcollectors(filenames))
+
+ driver = py.test.Driver(channel)
+ driver.run(fncollectors)
+
def getcollectors(filenames):
current = py.path.local()
- yielded = False
for fn in filenames:
fullfn = current.join(fn, abs=1)
if fullfn.check(file=1):
@@ -56,13 +120,10 @@
elif fullfn.check(dir=1):
yield py.test.collect.Directory(fullfn)
else:
- raise RuntimeError, "%r does not exist" % fn
- yielded = True
- if not yielded:
- yield py.test.collect.Directory(py.path.local())
+ raise IOError, "%r does not exist" % fn
def getanchors(args):
- """ yield anchors from skimming the args for existing files/dirs. """
+ """ yield "anchors" from skimming the args for existing files/dirs. """
current = py.path.local()
l = []
for arg in args:
Modified: py/dist/py/test/collect.py
==============================================================================
--- py/dist/py/test/collect.py (original)
+++ py/dist/py/test/collect.py Sun Oct 24 16:42:27 2004
@@ -23,6 +23,7 @@
Collector instances during iteration.
"""
Item = test.Item
+ Error = Error
def iterunits(self):
""" yield all units of the Collector instance. """
@@ -86,9 +87,9 @@
# we want to sort according to lineno, so here goes
# the generator lazyness
l = []
- for pypath in self.extpy.listdir():
+ for extpy in self.extpy.listdir():
for meth in self.yielders:
- for x in meth(pypath):
+ for x in meth(extpy):
x.fspath = self.extpy.root
sortvalue = self.getsortvalue(x)
l.append((sortvalue, x))
@@ -103,7 +104,7 @@
the same order as int he file.
"""
if isinstance(obj, self.Item):
- obj = obj.pypath.resolve()
+ obj = obj.extpy.resolve()
elif isinstance(obj, PyCollector):
for x in obj:
return self.getsortvalue(x)
@@ -128,24 +129,24 @@
except:
yield self._except()
- def collect_function(self, pypath):
- if pypath.check(func=1, basestarts='test_'):
- if self.extpy.samefile(pypath):
- yield self.Item(pypath)
-
- def collect_class(self, pypath):
- #print "checking %r (pypath: %r)" % (pypath.resolve(), pypath)
- if pypath.check(basestarts='Test') and self.extpy.samefile(pypath):
- obj = pypath.resolve()
+ def collect_function(self, extpy):
+ if extpy.check(func=1, basestarts='test_'):
+ if self.extpy.samefile(extpy):
+ yield self.Item(extpy)
+
+ def collect_class(self, extpy):
+ #print "checking %r (extpy: %r)" % (extpy.resolve(), extpy)
+ if extpy.check(basestarts='Test') and self.extpy.samefile(extpy):
+ obj = extpy.resolve()
if inspect.isclass(obj) and not getattr(obj, 'disabled', 0):
- yield Class(pypath)
+ yield Class(extpy)
class Class(PyCollector):
- def collect_method(self, pypath):
+ def collect_method(self, extpy):
# note that we want to allow inheritance and thus
# we don't check for "samemodule"-ness of test
# methods like in the Module Collector
- if pypath.check(basestarts='test_', func=1):
- func = pypath.resolve()
- yield getattr(func.im_class, 'Item', self.Item)(pypath)
+ if extpy.check(basestarts='test_', func=1):
+ func = extpy.resolve()
+ yield getattr(func.im_class, 'Item', self.Item)(extpy)
Modified: py/dist/py/test/compat.py
==============================================================================
--- py/dist/py/test/compat.py (original)
+++ py/dist/py/test/compat.py Sun Oct 24 16:42:27 2004
@@ -6,7 +6,7 @@
honouring setUp and tearDown semantics.
"""
def execute(self, runner):
- unboundmethod = self.pypath.resolve()
+ unboundmethod = self.extpy.resolve()
cls = unboundmethod.im_class
instance = cls()
instance.setUp()
@@ -14,7 +14,7 @@
unboundmethod(instance)
finally:
instance.tearDown()
- return py.test.Passed()
+ return py.test.Item.Passed()
class TestCase:
"""compatibility class of unittest's TestCase. """
@@ -28,7 +28,7 @@
def fail(self, msg=None):
""" fail immediate with given message. """
- raise py.test.Failed(msg=msg)
+ raise py.test.Item.Failed(msg=msg)
def assertRaises(self, excclass, func, *args, **kwargs):
py.test.raises(excclass, func, *args, **kwargs)
@@ -49,7 +49,7 @@
items.append("""
def %(name)s(self, %(sig)s):
if %(expr)s:
- raise py.test.Failed(tbindex=-2, msg=%(sigsubst)r %% (%(sig)s))
+ raise py.test.Item.Failed(tbindex=-2, msg=%(sigsubst)r %% (%(sig)s))
""" % locals() )
source = "".join(items)
Added: py/dist/py/test/item.py
==============================================================================
--- (empty file)
+++ py/dist/py/test/item.py Sun Oct 24 16:42:27 2004
@@ -0,0 +1,35 @@
+
+# ----------------------------------------------
+# Basic Test Item
+# ----------------------------------------------
+class Item(object):
+ _setupcache = []
+ _lastinstance = None
+
+ def __init__(self, extpy, *args):
+ self.extpy = extpy
+ self.name = extpy.basename
+ self.args = args
+
+ def execute(self, driver):
+ driver.setup_path(self.extpy)
+ target, teardown = driver.setup_method(self.extpy)
+ try:
+ target(*self.args)
+ finally:
+ if teardown:
+ teardown(target)
+
+ class Outcome:
+ def __init__(self, **kwargs):
+ assert 'msg' not in kwargs or isinstance(kwargs['msg'], str), (
+ "given 'msg' argument is not a string" )
+ self.__dict__.update(kwargs)
+ def __repr__(self):
+ return getattr(self, 'msg', object.__repr__(self))
+ class Passed(Outcome): pass
+ class Failed(Outcome): pass
+ class ExceptionFailure(Failed): pass
+ class Skipped(Outcome): pass
+
+
Modified: py/dist/py/test/raises.py
==============================================================================
--- py/dist/py/test/raises.py (original)
+++ py/dist/py/test/raises.py Sun Oct 24 16:42:27 2004
@@ -1,7 +1,6 @@
import sys
import py
-
-from py.__impl__ import test
+ExceptionFailure = py.test.Item.ExceptionFailure
def raises(ExpectedException, *args, **kwargs):
""" raise AssertionError, if target code does not raise the expected
@@ -25,8 +24,8 @@
excinfo = sys.exc_info()
else:
excinfo = None
- raise test.run.ExceptionFailure(expr=expr, expected=ExpectedException,
- innerexcinfo=excinfo, tbindex = -2)
+ raise ExceptionFailure(expr=expr, expected=ExpectedException,
+ innerexcinfo=excinfo, tbindex = -2)
else:
func = args[0]
assert callable
@@ -42,5 +41,5 @@
if k:
k = ', ' + k
expr = '%s(%r%s)' %(func.__name__, args, k)
- raise test.run.ExceptionFailure(expr=args, expected=ExpectedException,
- innerexcinfo=excinfo, tbindex = -2)
+ raise ExceptionFailure(expr=args, expected=ExpectedException,
+ innerexcinfo=excinfo, tbindex = -2)
Modified: py/dist/py/test/report/memo.py
==============================================================================
--- py/dist/py/test/report/memo.py (original)
+++ py/dist/py/test/report/memo.py Sun Oct 24 16:42:27 2004
@@ -1,10 +1,10 @@
from __future__ import generators
-import py
-from py.__impl__.test import run
+
+from py.test import Item
class MemoReporter:
typemap = {
- run.Passed: '.', run.Skipped: 's', run.Failed: 'F',
+ Item.Passed: '.', Item.Skipped: 's', Item.Failed: 'F',
}
def append(self, restype, testresult):
Modified: py/dist/py/test/report/text/reporter.py
==============================================================================
--- py/dist/py/test/report/text/reporter.py (original)
+++ py/dist/py/test/report/text/reporter.py Sun Oct 24 16:42:27 2004
@@ -1,21 +1,27 @@
from __future__ import generators
import py
+Item = py.test.Item
+Collector = py.test.collect.Collector
-from py.__impl__.test import run
from time import time as now
# lazy relative Implementation imports
from summary import Summary
from out import getout
+
class TextReporter(object):
Summary = Summary
typemap = {
- run.Passed: '.', run.Skipped: 's', run.Failed: 'F',
- py.test.collect.Error: 'C',
+ Item.Passed: '.',
+ Item.Skipped: 's',
+ Item.Failed: 'F',
+ Collector.Error: 'C',
}
namemap = {
- run.Passed: 'ok', run.Skipped: 'SKIP', run.Failed: 'FAIL',
- py.test.collect.Error: 'COLLECT ERROR',
+ Item.Passed: 'ok',
+ Item.Skipped: 'SKIP',
+ Item.Failed: 'FAIL',
+ Collector.Error: 'COLLECT ERROR',
}
def __init__(self, f=None):
@@ -66,7 +72,7 @@
numunits = len(list(collector.iterunits()))
if numunits > 0:
#if verbose < 2:
- # self.out.write('%s[%d]' % (collector.pypath,
+ # self.out.write('%s[%d]' % (collector.extpy,
# numunits))
#else:
#self.out.line("collector.fspy.modpath)
@@ -92,8 +98,8 @@
from py.__impl__.test.tool.outerrcapture import SimpleOutErrCapture
item.iocapture = SimpleOutErrCapture()
if self.out.tty:
- realpath, lineno = item.pypath.getfilelineno()
- location = "running %s:%d %s" % (realpath.basename, lineno, str(item.pypath.modpath))
+ realpath, lineno = item.extpy.getfilelineno()
+ location = "running %s:%d %s" % (realpath.basename, lineno, str(item.extpy.modpath))
self.out.rewrite(location)
self._started[item] = now()
@@ -111,9 +117,9 @@
restype, c = self.processresult(result)
writeinfo = None
if self.out.tty:
- if not isinstance(result, run.Passed) or self.option.verbose >=1:
+ if not isinstance(result, py.test.Item.Passed) or self.option.verbose >=1:
writeinfo = '\n'
- elif isinstance(result, run.Passed):
+ elif isinstance(result, py.test.Item.Passed):
writeinfo = ''
elif self.option.verbose >= 1:
writeinfo = '\n'
@@ -121,15 +127,15 @@
self.out.write(c)
if writeinfo is not None:
- realpath, lineno = item.pypath.getfilelineno()
+ realpath, lineno = item.extpy.getfilelineno()
location = "%s:%d" % (realpath.basename, lineno)
resultstring = self.namemap.get(restype, result.__class__.__name__)
self.out.rewrite("%.3f %-2s %-20s %s%s" % (
- elapsed, resultstring, location, str(item.pypath.modpath), writeinfo
+ elapsed, resultstring, location, str(item.extpy.modpath), writeinfo
))
if self.option.usepdb:
- if (issubclass(restype, py.test.collect.Error) or
- issubclass(restype, run.Failed)):
+ if (issubclass(restype, Collector.Error) or
+ issubclass(restype, Item.Failed)):
import pdb
self.out.rewrite(
'\n%s: %s\n'
@@ -137,17 +143,17 @@
result.excinfo[1]))
pdb.post_mortem(result.excinfo[2])
if self.option.exitfirstproblem:
- if (issubclass(restype, py.test.collect.Error) or
- issubclass(restype, run.Failed)):
- raise run.Exit("first problem, exit configured.")
+ if (issubclass(restype, Collector.Error) or
+ issubclass(restype, Item.Failed)):
+ py.test.exit("first problem, exit configured.")
def report_collect_error(self, error):
restype, c = self.processresult(error)
writeinfo = None
if self.out.tty:
- if not isinstance(result, run.Passed) or self.option.verbose >=1:
+ if not isinstance(result, Item.Passed) or self.option.verbose >=1:
writeinfo = '\n'
- elif isinstance(result, run.Passed):
+ elif isinstance(result, Item.Passed):
writeinfo = ''
elif self.option.verbose >= 1:
writeinfo = '\n'
Modified: py/dist/py/test/report/text/summary.py
==============================================================================
--- py/dist/py/test/report/text/summary.py (original)
+++ py/dist/py/test/report/text/summary.py Sun Oct 24 16:42:27 2004
@@ -1,6 +1,6 @@
from __future__ import generators
import py
-from py.__impl__.test import run
+Item = py.test.Item
from py.__impl__.magic import exprinfo, assertion
class Summary(object):
@@ -45,7 +45,7 @@
self.summary_collect_errors()
outlist = []
sum = 0
- for typ in run.Passed, run.Failed, run.Skipped:
+ for typ in Item.Passed, Item.Failed, Item.Skipped:
l = self.getlist(typ)
outlist.append('%d %s' % (len(l), typ.__name__.lower()))
sum += len(l)
@@ -62,8 +62,9 @@
def skippedreasons(self):
d = {}
- for res in self.getlist(run.Skipped):
- raisingtb = py.magic.dyncode.listtb(res.excinfo[2])[-1]
+ for res in self.getlist(Item.Skipped):
+ tbindex = getattr(res, 'tbindex', -1)
+ raisingtb = py.magic.dyncode.listtb(res.excinfo[2])[tbindex]
fn = raisingtb.tb_frame.f_code.co_filename
lineno = raisingtb.tb_lineno
d[(fn,lineno)] = res
@@ -75,7 +76,7 @@
self.out.line()
def failures(self):
- for res in self.getlist(run.Failed):
+ for res in self.getlist(Item.Failed):
self.repr_failure(res)
def repr_failure(self, res):
@@ -83,7 +84,7 @@
self.out.line()
self.out.sep("_")
self.out.line()
- #self.out.sep("_", "Test Failure") # %s" % res.unit.pypath)
+ #self.out.sep("_", "Test Failure") # %s" % res.unit.extpy)
#self.out.sep("_")
#self.out.line()
self.repr_traceback(res.item, res.excinfo[2],
@@ -105,7 +106,7 @@
def repr_failure_result(self, res):
cls = res.excinfo[0]
- if issubclass(cls, run.ExceptionFailure):
+ if issubclass(cls, Item.ExceptionFailure):
if not res.innerexcinfo:
self.out.line("%s <<< DID NOT RAISE" % (res.expr,))
self.out.line("expected: %r" % (res.expected,))
@@ -184,14 +185,14 @@
return old
def repr_traceback(self, item, tb, tbindex=-1):
- t_file, t_lineno = item.pypath.getfilelineno()
- self.out.line("%s, line %d" % (item.pypath, t_lineno) )
- fspath = item.pypath.root
+ t_file, t_lineno = item.extpy.getfilelineno()
+ self.out.line("%s, line %d" % (item.extpy, t_lineno) )
+ fspath = item.extpy.root
self.out.sep('_')
self.repr_traceback_raw(fspath, tb, tbindex)
- #t_file, t_lineno = unit.pypath.getfilelineno()
+ #t_file, t_lineno = unit.extpy.getfilelineno()
#if t_file != filename or lineno != t_lineno:
- #self.out.line("%s, line %d" % (unit.pypath, t_lineno) )
+ #self.out.line("%s, line %d" % (unit.extpy, t_lineno) )
self.out.sep('-')
def repr_traceback_raw(self, fspath, tb, tbindex=-1):
Modified: py/dist/py/test/run.py
==============================================================================
--- py/dist/py/test/run.py (original)
+++ py/dist/py/test/run.py Sun Oct 24 16:42:27 2004
@@ -1,33 +1,58 @@
from __future__ import generators
import py
+collect = py.test.collect
class Exit(Exception):
- """ for immediate program exits without tracebacks. """
+ """ for immediate program exits without tracebacks and reporter/summary. """
+
+def exit(*args):
+ raise Exit(*args)
+
+def skip(msg="unknown reason"):
+ """ skip with the given Message. """
+ raise py.test.Item.Skipped(msg=msg, tbindex=-2)
+
+def fail(msg="unknown failure"):
+ """ fail with the given Message. """
+ raise py.test.Item.Failed(msg=msg, tbindex=-2)
class Driver:
option = py.test.config.option
Exit = Exit
- def __init__(self, reporter):
- self.reporter = reporter
+ def __init__(self, channel):
+ self.reporter = py.test.config.getfirst('getreporter') ()
self._setupstack = []
self._instance = None
+ self._channel = channel
- def run(self, obj):
- raise ValueError
+ def run(self, collectors):
+ """ main loop for running tests. """
+ self.setup()
+ try:
+ self.reporter.start()
+ try:
+ for x in collectors:
+ self.run_collector_or_item(x)
+ except self.Exit:
+ pass
+ finally:
+ self.teardown()
+ self.reporter.end()
def run_collector_or_item(self, obj):
""" run (possibly many) testitems and/or collectors. """
- collect = py.test.collect
- if isinstance(obj, Item):
+ if self._channel.isclosed():
+ raise SystemExit, "Channel is closed"
+ if isinstance(obj, py.test.Item):
if not self.option.collectonly:
#if self.option.args:
# if str(obj.path).find(self.option.args[0]) == -1:
# return
self.runitem(obj)
- elif isinstance(obj, collect.Collector):
+ elif isinstance(obj, py.test.collect.Collector):
self.runcollector(obj)
- elif isinstance(obj, collect.Error):
+ elif isinstance(obj, py.test.collect.Collector.Error):
try:
self.reporter.report_collect_error(obj)
except (KeyboardInterrupt, SystemExit):
@@ -56,41 +81,47 @@
def runitem(self, item):
self.reporter.startitem(item)
try:
- res = item.execute(self) or Passed()
- except Outcome, res:
+ res = item.execute(self) or item.Passed()
+ except item.Outcome, res:
res.excinfo = py.std.sys.exc_info()
except (KeyboardInterrupt, SystemExit):
raise
except:
- res = Failed(excinfo=py.std.sys.exc_info())
+ res = item.Failed(excinfo=py.std.sys.exc_info())
+ if not isinstance(res, item.Passed):
+ self._failed.append(item)
res.item = item
self.reporter.enditem(res)
- def setup_path(self, pypath):
+ def setup_path(self, extpy):
""" setup objects along the path to the test-method
- (pointed to by pypath). Tear down any previously
+ (pointed to by extpy). Tear down any previously
setup objects which are not directly needed.
"""
- # setupstack contains (pypath, obj)'s of already setup objects
- # strict ordering is maintained, i.e. each pypath in
- # the stack is "relto" the previous pypath.
+ # setupstack contains (extpy, obj)'s of already setup objects
+ # strict ordering is maintained, i.e. each extpy in
+ # the stack is "relto" the previous extpy.
stack = self._setupstack
- while stack and not pypath.relto(stack[-1][0]):
+ while stack and not extpy.relto(stack[-1][0]):
self._teardownone(stack.pop()[1])
- rest = pypath.parts()[len(stack):-1]
+ rest = extpy.parts()[len(stack):-1]
for x in rest:
stack.append((x, self._setupone(x)))
+ def getfailed(self):
+ return self._failed
+
def setup(self):
""" setup any neccessary resources. """
+ self._failed = []
def teardown(self):
""" teardown any resources the driver knows about. """
while self._setupstack:
self._teardownone(self._setupstack.pop()[1])
- def _setupone(self, pypath):
- obj = pypath.resolve()
+ def _setupone(self, extpy):
+ obj = extpy.resolve()
if py.std.inspect.ismodule(obj):
if hasattr(obj, 'setup_module'):
obj.setup_module(obj)
@@ -107,9 +138,9 @@
if hasattr(obj, 'teardown_class'):
obj.teardown_class.im_func(obj)
- def setup_method(self, pypath):
+ def setup_method(self, extpy):
""" return a tuple of (bound method or callable, teardown method). """
- method = pypath.resolve()
+ method = extpy.resolve()
if not hasattr(method, 'im_class'):
return method, None
if self._instance.__class__ != method.im_class:
@@ -119,38 +150,3 @@
print "execting setup", self._instance.setup_method
self._instance.setup_method(method)
return (method, getattr(self._instance, 'teardown_method', None))
-
-# ----------------------------------------------
-# Basic Test Unit Executor
-# ----------------------------------------------
-class Item(object):
- _setupcache = []
- _lastinstance = None
-
- def __init__(self, pypath, *args):
- self.pypath = pypath
- self.name = pypath.basename
- self.args = args
-
- def execute(self, driver):
- driver.setup_path(self.pypath)
- target, teardown = driver.setup_method(self.pypath)
- try:
- target(*self.args)
- finally:
- if teardown:
- teardown(target)
-
-class Outcome:
- def __init__(self, **kwargs):
- assert 'msg' not in kwargs or isinstance(kwargs['msg'], str), (
- "given 'msg' argument is not a string" )
- self.__dict__.update(kwargs)
- def __repr__(self):
- return getattr(self, 'msg', object.__repr__(self))
-class Passed(Outcome): pass
-class Failed(Outcome): pass
-class ExceptionFailure(Failed): pass
-class Skipped(Outcome): pass
-
-
Modified: py/dist/py/test/test_compat.py
==============================================================================
--- py/dist/py/test/test_compat.py (original)
+++ py/dist/py/test/test_compat.py Sun Oct 24 16:42:27 2004
@@ -44,7 +44,7 @@
#self.%(name)s(%(paramfail)s)
def test_%(name)s_failing(self):
- self.assertRaises(py.test.Failed,
+ self.assertRaises(py.test.Item.Failed,
self.%(name)s, %(paramfail)s)
""" % locals()
co = py.magic.dyncode.compile2(source)
More information about the pytest-commit
mailing list