[py-svn] r33320 - in py/dist/py/test: rsession rsession/testing tracer
fijal at codespeak.net
fijal at codespeak.net
Mon Oct 16 12:29:13 CEST 2006
Author: fijal
Date: Mon Oct 16 12:28:23 2006
New Revision: 33320
Added:
py/dist/py/test/rsession/local.py (contents, props changed)
py/dist/py/test/rsession/testing/test_lsession.py (contents, props changed)
Modified:
py/dist/py/test/rsession/rsession.py
py/dist/py/test/tracer/docstorage.py
Log:
Added first quasi-working version of local-version of rsession. (reporter needs
refactoring for sure to work with different outcomes).
Added: py/dist/py/test/rsession/local.py
==============================================================================
--- (empty file)
+++ py/dist/py/test/rsession/local.py Mon Oct 16 12:28:23 2006
@@ -0,0 +1,22 @@
+
+""" local-only operations
+"""
+
+from py.__.test.rsession.executor import RunExecutor
+from py.__.test.rsession import report
+
+def run(item):
+ r = RunExecutor(item)
+ return r.execute()
+
+def local_loop(reporter, itemgenerator, shouldstop):
+
+ while 1:
+ try:
+ item = itemgenerator.next()
+ if shouldstop():
+ return
+ outcome = run(item)
+ reporter(report.ReceivedItemOutcome(None, item, outcome))
+ except StopIteration:
+ break
Modified: py/dist/py/test/rsession/rsession.py
==============================================================================
--- py/dist/py/test/rsession/rsession.py (original)
+++ py/dist/py/test/rsession/rsession.py Mon Oct 16 12:28:23 2006
@@ -16,6 +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
class RemoteOptions(object):
def __init__(self, d):
@@ -224,14 +225,79 @@
def report_Nodes(self, event):
self.nodes = event.nodes
-class RSession(object):
+class AbstractSession(object):
"""
- An RSession executes collectors/items through a runner.
+ An abstract session executes collectors/items through a runner.
"""
def __init__(self, config):
self.config = config
+
+ def make_colitems(paths, baseon):
+ # we presume that from the base we can simply get to
+ # the target paths by joining the basenames
+ res = []
+ for x in paths:
+ x = py.path.local(x)
+ current = py.test.collect.Directory(baseon)
+ relparts = x.relto(baseon).split(x.sep)
+ assert relparts
+ for part in relparts:
+ next = current.join(part)
+ assert next is not None, (current, part)
+ current = next
+ res.append(current)
+ return res
+ make_colitems = staticmethod(make_colitems)
+
+ def getpkgdir(path):
+ path = py.path.local(path)
+ pkgpath = path.pypkgpath()
+ if pkgpath is None:
+ pkgpath = path
+ return pkgpath
+ getpkgdir = staticmethod(getpkgdir)
+
+ def init_reporter(self, reporter, sshhosts):
+ try:
+ # XXX: use it like a command line option, but how?
+ startserverflag = self.config.getinitialvalue("startserver")
+ except:
+ startserverflag = False
+
+ checkfun = lambda: None
+ if startserverflag and reporter is None:
+ from py.__.test.rsession.web import start_server, exported_methods
+
+ reporter = exported_methods.report
+ start_server()
+ elif reporter is None:
+ reporter_instance = Reporter(self.config, sshhosts)
+ reporter = reporter_instance.report
+ checkfun = lambda : self.config.option.exitfirst and \
+ reporter_instance.is_failing()
+ else:
+ startserverflag = False
+ if reporter is None:
+ reporter_instance = Reporter(self.config, sshhosts)
+ reporter = reporter_instance.report
+ checkfun = lambda : self.config.option.exitfirst and \
+ reporter_instance.is_failing()
+ return reporter, checkfun, startserverflag
+
+ def reporterror(reporter, data):
+ excinfo, item = data
+ if excinfo is None:
+ reporter(report.ItemStart(item))
+ elif excinfo.type is py.test.Item.Skipped:
+ reporter(report.SkippedTryiter(excinfo, item))
+ else:
+ reporter(report.FailedTryiter(excinfo, item))
+ reporterror = staticmethod(reporterror)
+class RSession(AbstractSession):
+ """ Remote version of session
+ """
def main(self, args, reporter=None):
""" main loop for running tests. """
if not args:
@@ -251,25 +317,9 @@
rsync_roots = self.config.getinitialvalue("distrsync_roots")
except:
rsync_roots = None # all files and directories in the pkgdir
- try:
- # XXX: use it like a command line option, but how?
- startserverflag = self.config.getinitialvalue("startserver")
- except:
- startserverflag = False
- checkfun = lambda: None
- if startserverflag and reporter is None:
- from py.__.test.rsession.web import start_server, exported_methods
-
- reporter = exported_methods.report
- start_server()
- elif reporter is None:
- reporter_instance = Reporter(self.config, sshhosts)
- reporter = reporter_instance.report
- checkfun = lambda : self.config.option.exitfirst and \
- reporter_instance.is_failing()
- else:
- startserverflag = False
+ reporter, checkfun, startserverflag = self.init_reporter(reporter,
+ sshhosts)
reporter(report.TestStarted(sshhosts))
pkgdir = self.getpkgdir(args[0])
colitems = self.make_colitems(args, baseon=pkgdir.dirpath())
@@ -282,20 +332,11 @@
rsync_roots, remotepython)
reporter(report.RsyncFinished())
- def reporterror(data):
- excinfo, item = data
- if excinfo is None:
- reporter(report.ItemStart(item))
- elif excinfo.type is py.test.Item.Skipped:
- reporter(report.SkippedTryiter(excinfo, item))
- else:
- reporter(report.FailedTryiter(excinfo, item))
-
keyword = self.config.option.keyword
def itemgen():
for x in colitems:
- for y in x.tryiter(reporterror = reporterror, keyword = keyword):
+ for y in x.tryiter(reporterror = lambda x: self.reporterror(reporter, x), keyword = keyword):
yield y
itemgenerator = itemgen()
@@ -308,28 +349,37 @@
if startserverflag:
from py.__.test.rsession.web import kill_server
kill_server()
+
+class LSession(AbstractSession):
+ """ Local version of session
+ """
+ def main(self, args, reporter=None):
+ if not args:
+ args = [py.path.local()]
+
+ sshhosts = ['localhost'] # this is just an info to reporter
+
+ reporter, checkfun, startserverflag = self.init_reporter(reporter,
+ sshhosts)
- def make_colitems(paths, baseon):
- # we presume that from the base we can simply get to
- # the target paths by joining the basenames
- res = []
- for x in paths:
- x = py.path.local(x)
- current = py.test.collect.Directory(baseon)
- relparts = x.relto(baseon).split(x.sep)
- assert relparts
- for part in relparts:
- next = current.join(part)
- assert next is not None, (current, part)
- current = next
- res.append(current)
- return res
- make_colitems = staticmethod(make_colitems)
+ reporter(report.TestStarted(sshhosts))
+ pkgdir = self.getpkgdir(args[0])
+ colitems = self.make_colitems(args, baseon=pkgdir.dirpath())
+ reporter(report.RsyncFinished())
- def getpkgdir(path):
- path = py.path.local(path)
- pkgpath = path.pypkgpath()
- if pkgpath is None:
- pkgpath = path
- return pkgpath
- getpkgdir = staticmethod(getpkgdir)
+ keyword = self.config.option.keyword
+
+ def itemgen():
+ for x in colitems:
+ for y in x.tryiter(reporterror = lambda x: self.reporterror(reporter, x), keyword = keyword):
+ yield y
+
+ itemgenerator = itemgen()
+ #assert 0, "\n".join([",".join(x.listnames()) for x in
+ # list(itemgenerator)])
+ local_loop(reporter, itemgenerator, checkfun)
+
+ reporter(report.TestFinished())
+ if startserverflag:
+ from py.__.test.rsession.web import kill_server
+ kill_server()
Added: py/dist/py/test/rsession/testing/test_lsession.py
==============================================================================
--- (empty file)
+++ py/dist/py/test/rsession/testing/test_lsession.py Mon Oct 16 12:28:23 2006
@@ -0,0 +1,49 @@
+
+""" test of local version of py.test distributed
+"""
+
+import py
+from py.__.test.rsession.rsession import LSession
+from py.__.test.rsession import report
+
+class TestLSession(object):
+ def test_example_distribution(self):
+ # XXX find a better way for the below
+ tmpdir = py.path.local(py.__file__).dirpath().dirpath()
+ tmpdir.ensure("sub", "__init__.py")
+ tmpdir.ensure("sub", "test_one.py").write(py.code.Source("""
+ def test_1():
+ pass
+ def test_2():
+ assert 0
+ def test_3():
+ raise ValueError(23)
+ def test_4(someargs):
+ pass
+ """))
+ args = [str(tmpdir.join("sub"))]
+ config, args = py.test.Config.parse(args)
+ lsession = LSession(config)
+ allevents = []
+ lsession.main(args, reporter=allevents.append)
+ 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]
+ assert len(passevents) == 1
+ assert len(failevents) == 3
+ tb = failevents[0].outcome.excinfo.traceback
+ assert str(tb[0].path).find("test_one") != -1
+ assert str(tb[0].getsource()).find("test_2") != -1
+ assert failevents[0].outcome.excinfo.type is AssertionError
+ tb = failevents[1].outcome.excinfo.traceback
+ assert str(tb[0].path).find("test_one") != -1
+ assert str(tb[0].getsource()).find("test_3") != -1
+ assert failevents[1].outcome.excinfo.type is ValueError
+ assert str(failevents[1].outcome.excinfo.value) == '23'
+ tb = failevents[2].outcome.excinfo.traceback
+ assert failevents[2].outcome.excinfo.type is TypeError
+ assert str(tb[0].path).find("executor") != -1
+ assert str(tb[0].getsource()).find("execute") != -1
Modified: py/dist/py/test/tracer/docstorage.py
==============================================================================
--- py/dist/py/test/tracer/docstorage.py (original)
+++ py/dist/py/test/tracer/docstorage.py Mon Oct 16 12:28:23 2006
@@ -13,8 +13,8 @@
try:
from pypy.annotation.policy import AnnotatorPolicy
from pypy.annotation.bookkeeper import Bookkeeper
-except ImportError:
- py.test.skip("No PyPy")
+except ImportError, i:
+ py.test.skip("No PyPy %s" % i)
class DummyAnnotator(object):
policy = AnnotatorPolicy()
More information about the pytest-commit
mailing list