[py-svn] r46011 - in py/trunk/py/test: . rsession rsession/testing rsession/webdata testing
fijal at codespeak.net
fijal at codespeak.net
Mon Aug 27 11:02:51 CEST 2007
Author: fijal
Date: Mon Aug 27 11:02:50 2007
New Revision: 46011
Added:
py/trunk/py/test/repevent.py
- copied unchanged from r46010, py/branch/session-cleanups/py/test/repevent.py
py/trunk/py/test/reporter.py
- copied unchanged from r46010, py/branch/session-cleanups/py/test/reporter.py
py/trunk/py/test/testing/test_outcome.py
- copied unchanged from r46010, py/branch/session-cleanups/py/test/testing/test_outcome.py
py/trunk/py/test/testing/test_repevent.py
- copied unchanged from r46010, py/branch/session-cleanups/py/test/testing/test_repevent.py
py/trunk/py/test/testing/test_reporter.py
- copied unchanged from r46010, py/branch/session-cleanups/py/test/testing/test_reporter.py
Removed:
py/trunk/py/test/rsession/outcome.py
py/trunk/py/test/rsession/repevent.py
py/trunk/py/test/rsession/reporter.py
py/trunk/py/test/rsession/testing/test_outcome.py
py/trunk/py/test/rsession/testing/test_repevent.py
py/trunk/py/test/rsession/testing/test_reporter.py
Modified:
py/trunk/py/test/collect.py
py/trunk/py/test/outcome.py
py/trunk/py/test/rsession/executor.py
py/trunk/py/test/rsession/hostmanage.py
py/trunk/py/test/rsession/local.py
py/trunk/py/test/rsession/master.py
py/trunk/py/test/rsession/rest.py
py/trunk/py/test/rsession/rsession.py
py/trunk/py/test/rsession/slave.py
py/trunk/py/test/rsession/testing/test_executor.py
py/trunk/py/test/rsession/testing/test_hostmanage.py
py/trunk/py/test/rsession/testing/test_lsession.py
py/trunk/py/test/rsession/testing/test_master.py
py/trunk/py/test/rsession/testing/test_rest.py
py/trunk/py/test/rsession/testing/test_rsession.py
py/trunk/py/test/rsession/testing/test_slave.py
py/trunk/py/test/rsession/web.py
py/trunk/py/test/rsession/webdata/source.js
py/trunk/py/test/session.py
py/trunk/py/test/testing/test_collect.py
py/trunk/py/test/testing/test_config.py
Log:
Branch merge of session cleanups.
* killed _tryiter usage in rsession
* moved reporter one level up, so it can be reused later for normal session
* a lot of small simplifications
Modified: py/trunk/py/test/collect.py
==============================================================================
--- py/trunk/py/test/collect.py (original)
+++ py/trunk/py/test/collect.py Mon Aug 27 11:02:50 2007
@@ -189,7 +189,7 @@
return True
return False
- def _tryiter(self, yieldtype=None, reporterror=None, keyword=None):
+ def _tryiter(self, yieldtype=None):
""" yield stop item instances from flattening the collector.
XXX deprecated: this way of iteration is not safe in all
cases.
@@ -197,28 +197,17 @@
if yieldtype is None:
yieldtype = py.test.collect.Item
if isinstance(self, yieldtype):
- try:
- self._skipbykeyword(keyword)
- yield self
- except Skipped:
- if reporterror is not None:
- excinfo = py.code.ExceptionInfo()
- reporterror((excinfo, self))
+ yield self
else:
if not isinstance(self, py.test.collect.Item):
try:
- if reporterror is not None:
- reporterror((None, self))
for x in self.run():
- for y in self.join(x)._tryiter(yieldtype,
- reporterror, keyword):
+ for y in self.join(x)._tryiter(yieldtype):
yield y
except KeyboardInterrupt:
raise
- except:
- if reporterror is not None:
- excinfo = py.code.ExceptionInfo()
- reporterror((excinfo, self))
+ except:
+ pass
def _getsortvalue(self):
return self.name
Modified: py/trunk/py/test/outcome.py
==============================================================================
--- py/trunk/py/test/outcome.py (original)
+++ py/trunk/py/test/outcome.py Mon Aug 27 11:02:50 2007
@@ -1,7 +1,10 @@
-""" File defining possible outcomes of running
+""" File defining possible outcomes of running and also
+serialization of outcomes
"""
+import py, sys
+
class Outcome:
def __init__(self, msg=None, excinfo=None):
self.msg = msg
@@ -27,3 +30,120 @@
class Skipped(Outcome):
pass
+
+
+class SerializableOutcome(object):
+ def __init__(self, setupfailure=False, excinfo=None, skipped=None,
+ is_critical=False):
+ self.passed = not excinfo and not skipped
+ self.skipped = skipped
+ self.setupfailure = setupfailure
+ self.excinfo = excinfo
+ self.is_critical = is_critical
+ self.signal = 0
+ self.stdout = "" # XXX temporary
+ self.stderr = ""
+ assert bool(self.passed) + bool(excinfo) + bool(skipped) == 1
+
+ def make_excinfo_repr(self, tbstyle):
+ if self.excinfo is None:
+ return None
+ excinfo = self.excinfo
+ tb_info = [self.traceback_entry_repr(x, tbstyle)
+ for x in excinfo.traceback]
+ rec_index = excinfo.traceback.recursionindex()
+ if hasattr(excinfo, 'type'):
+ etype = excinfo.type
+ if hasattr(etype, '__name__'):
+ etype = etype.__name__
+ else:
+ etype = excinfo.typename
+ val = getattr(excinfo, 'value', None)
+ if not val:
+ val = excinfo.exconly()
+ val = str(val)
+ return (etype, val, (tb_info, rec_index))
+
+ def traceback_entry_repr(self, tb_entry, tb_style):
+ lineno = tb_entry.lineno
+ relline = lineno - tb_entry.frame.code.firstlineno
+ path = str(tb_entry.path)
+ #try:
+ try:
+ if tb_style == 'long':
+ source = str(tb_entry.getsource())
+ else:
+ source = str(tb_entry.getsource()).split("\n")[relline]
+ except py.error.ENOENT:
+ source = "[cannot get source]"
+ name = str(tb_entry.frame.code.name)
+ # XXX: Bare except. What can getsource() raise anyway?
+ # SyntaxError, AttributeError, IndentationError for sure, check it
+ #except:
+ # source = "<could not get source>"
+ return (relline, lineno, source, path, name)
+
+ def make_repr(self, tbstyle="long"):
+ return (self.passed, self.setupfailure,
+ self.make_excinfo_repr(tbstyle),
+ self.skipped, self.is_critical, 0, self.stdout, self.stderr)
+
+class TracebackEntryRepr(object):
+ def __init__(self, tbentry):
+ relline, lineno, self.source, self.path, self.name = tbentry
+ self.relline = int(relline)
+ self.path = py.path.local(self.path)
+ self.lineno = int(lineno)
+ self.locals = {}
+
+ def __repr__(self):
+ return "line %s in %s\n %s" %(self.lineno, self.path, self.source[100:])
+
+ def getsource(self):
+ return py.code.Source(self.source).strip()
+
+ def getfirstlinesource(self):
+ return self.lineno - self.relline
+
+class TracebackRepr(list):
+ def recursionindex(self):
+ return self.recursion_index
+
+class ExcInfoRepr(object):
+ def __init__(self, excinfo):
+ self.typename, self.value, tb_i = excinfo
+ tb, rec_index = tb_i
+ self.traceback = TracebackRepr([TracebackEntryRepr(x) for x in tb])
+ self.traceback.recursion_index = rec_index
+
+ def __repr__(self):
+ l = ["%s=%s" %(x, getattr(self, x))
+ for x in "typename value traceback".split()]
+ return "<ExcInfoRepr %s>" %(" ".join(l),)
+
+ def exconly(self, tryshort=False):
+ """ Somehow crippled version of original one
+ """
+ return "%s: %s" % (self.typename, self.value)
+
+ def errisinstance(self, exc_t):
+ if not isinstance(exc_t, tuple):
+ exc_t = (exc_t,)
+ for exc in exc_t:
+ if self.typename == str(exc).split('.')[-1]:
+ return True
+ return False
+
+class ReprOutcome(object):
+ def __init__(self, repr_tuple):
+ (self.passed, self.setupfailure, excinfo, self.skipped,
+ self.is_critical, self.signal, self.stdout, self.stderr) = repr_tuple
+ if excinfo is None:
+ self.excinfo = None
+ else:
+ self.excinfo = ExcInfoRepr(excinfo)
+
+ def __repr__(self):
+ l = ["%s=%s" %(x, getattr(self, x))
+ for x in "signal passed skipped setupfailure excinfo stdout stderr".split()]
+ return "<ReprOutcome %s>" %(" ".join(l),)
Modified: py/trunk/py/test/rsession/executor.py
==============================================================================
--- py/trunk/py/test/rsession/executor.py (original)
+++ py/trunk/py/test/rsession/executor.py Mon Aug 27 11:02:50 2007
@@ -3,9 +3,9 @@
import py, os, sys
-from py.__.test.rsession.outcome import Outcome, ReprOutcome
+from py.__.test.outcome import SerializableOutcome, ReprOutcome
from py.__.test.rsession.box import Box
-from py.__.test.rsession import repevent
+from py.__.test import repevent
from py.__.test.outcome import Skipped, Failed
class RunExecutor(object):
@@ -33,9 +33,9 @@
def execute(self, capture=True):
try:
self.run(capture)
- outcome = Outcome()
+ outcome = SerializableOutcome()
except Skipped, e:
- outcome = Outcome(skipped=str(e))
+ outcome = SerializableOutcome(skipped=str(e))
except (SystemExit, KeyboardInterrupt):
raise
except:
@@ -49,7 +49,7 @@
code = py.code.Code(fun)
excinfo.traceback = excinfo.traceback.cut(
path=code.path, firstlineno=code.firstlineno)
- outcome = Outcome(excinfo=excinfo, setupfailure=False)
+ outcome = SerializableOutcome(excinfo=excinfo, setupfailure=False)
if self.usepdb:
if self.reporter is not None:
self.reporter(repevent.ImmediateFailure(self.item,
Modified: py/trunk/py/test/rsession/hostmanage.py
==============================================================================
--- py/trunk/py/test/rsession/hostmanage.py (original)
+++ py/trunk/py/test/rsession/hostmanage.py Mon Aug 27 11:02:50 2007
@@ -5,7 +5,7 @@
from py.__.test.rsession.master import MasterNode
from py.__.test.rsession.slave import setup_slave
-from py.__.test.rsession import repevent
+from py.__.test import repevent
class HostInfo(object):
""" Class trying to store all necessary attributes
Modified: py/trunk/py/test/rsession/local.py
==============================================================================
--- py/trunk/py/test/rsession/local.py (original)
+++ py/trunk/py/test/rsession/local.py Mon Aug 27 11:02:50 2007
@@ -5,8 +5,8 @@
import py
from py.__.test.rsession.executor import BoxExecutor, RunExecutor,\
ApigenExecutor
-from py.__.test.rsession import repevent
-from py.__.test.rsession.outcome import ReprOutcome
+from py.__.test import repevent
+from py.__.test.outcome import ReprOutcome
# XXX copied from session.py
def startcapture(session):
Modified: py/trunk/py/test/rsession/master.py
==============================================================================
--- py/trunk/py/test/rsession/master.py (original)
+++ py/trunk/py/test/rsession/master.py Mon Aug 27 11:02:50 2007
@@ -2,8 +2,9 @@
Node code for Master.
"""
import py
-from py.__.test.rsession.outcome import ReprOutcome
-from py.__.test.rsession import repevent
+from py.__.test.outcome import ReprOutcome
+from py.__.test import repevent
+from py.__.test.outcome import Skipped
class MasterNode(object):
def __init__(self, channel, reporter):
@@ -39,12 +40,30 @@
# of hanging nodes and such
raise
-def itemgen(colitems, reporter, keyword, reporterror):
- def rep(x):
- reporterror(reporter, x)
- for x in colitems:
- for y in x._tryiter(reporterror=rep, keyword=keyword):
- yield y
+def itemgen(colitems, reporter, keyword=None):
+ stopitems = py.test.collect.Item # XXX should be generator here as well
+ for next in colitems:
+ if isinstance(next, stopitems):
+ try:
+ next._skipbykeyword(keyword)
+ yield next
+ except Skipped:
+ excinfo = py.code.ExceptionInfo()
+ reporter(repevent.SkippedTryiter(excinfo, next))
+ else:
+ reporter(repevent.ItemStart(next))
+ try:
+ for x in itemgen([next.join(x) for x in next.run()], reporter,
+ keyword):
+ yield x
+ except (KeyboardInterrupt, SystemExit, GeneratorExit):
+ raise
+ except:
+ excinfo = py.code.ExceptionInfo()
+ if excinfo.type is Skipped:
+ reporter(repevent.SkippedTryiter(excinfo, next))
+ else:
+ reporter(repevent.FailedTryiter(excinfo, next))
def dispatch_loop(masternodes, itemgenerator, shouldstop,
waiter = lambda: py.std.time.sleep(0.1),
Deleted: /py/trunk/py/test/rsession/outcome.py
==============================================================================
--- /py/trunk/py/test/rsession/outcome.py Mon Aug 27 11:02:50 2007
+++ (empty file)
@@ -1,126 +0,0 @@
-
-""" Classes for representing outcomes on master and slavenode sides
-"""
-
-# WARNING! is_critical is debugging flag which means something
-# wrong went on a different level. Usually that means
-# internal bug.
-
-import sys
-import py
-
-class Outcome(object):
- def __init__(self, setupfailure=False, excinfo=None, skipped=None,
- is_critical=False):
- self.passed = not excinfo and not skipped
- self.skipped = skipped
- self.setupfailure = setupfailure
- self.excinfo = excinfo
- self.is_critical = is_critical
- self.signal = 0
- self.stdout = "" # XXX temporary
- self.stderr = ""
- assert bool(self.passed) + bool(excinfo) + bool(skipped) == 1
-
- def make_excinfo_repr(self, tbstyle):
- if self.excinfo is None:
- return None
- excinfo = self.excinfo
- tb_info = [self.traceback_entry_repr(x, tbstyle)
- for x in excinfo.traceback]
- rec_index = excinfo.traceback.recursionindex()
- if hasattr(excinfo, 'type'):
- etype = excinfo.type
- if hasattr(etype, '__name__'):
- etype = etype.__name__
- else:
- etype = excinfo.typename
- val = getattr(excinfo, 'value', None)
- if not val:
- val = excinfo.exconly()
- val = str(val)
- return (etype, val, (tb_info, rec_index))
-
- def traceback_entry_repr(self, tb_entry, tb_style):
- lineno = tb_entry.lineno
- relline = lineno - tb_entry.frame.code.firstlineno
- path = str(tb_entry.path)
- #try:
- try:
- if tb_style == 'long':
- source = str(tb_entry.getsource())
- else:
- source = str(tb_entry.getsource()).split("\n")[relline]
- except py.error.ENOENT:
- source = "[cannot get source]"
- name = str(tb_entry.frame.code.name)
- # XXX: Bare except. What can getsource() raise anyway?
- # SyntaxError, AttributeError, IndentationError for sure, check it
- #except:
- # source = "<could not get source>"
- return (relline, lineno, source, path, name)
-
- def make_repr(self, tbstyle="long"):
- return (self.passed, self.setupfailure,
- self.make_excinfo_repr(tbstyle),
- self.skipped, self.is_critical, 0, self.stdout, self.stderr)
-
-class TracebackEntryRepr(object):
- def __init__(self, tbentry):
- relline, lineno, self.source, self.path, self.name = tbentry
- self.relline = int(relline)
- self.path = py.path.local(self.path)
- self.lineno = int(lineno)
- self.locals = {}
-
- def __repr__(self):
- return "line %s in %s\n %s" %(self.lineno, self.path, self.source[100:])
-
- def getsource(self):
- return py.code.Source(self.source).strip()
-
- def getfirstlinesource(self):
- return self.lineno - self.relline
-
-class TracebackRepr(list):
- def recursionindex(self):
- return self.recursion_index
-
-class ExcInfoRepr(object):
- def __init__(self, excinfo):
- self.typename, self.value, tb_i = excinfo
- tb, rec_index = tb_i
- self.traceback = TracebackRepr([TracebackEntryRepr(x) for x in tb])
- self.traceback.recursion_index = rec_index
-
- def __repr__(self):
- l = ["%s=%s" %(x, getattr(self, x))
- for x in "typename value traceback".split()]
- return "<ExcInfoRepr %s>" %(" ".join(l),)
-
- def exconly(self, tryshort=False):
- """ Somehow crippled version of original one
- """
- return "%s: %s" % (self.typename, self.value)
-
- def errisinstance(self, exc_t):
- if not isinstance(exc_t, tuple):
- exc_t = (exc_t,)
- for exc in exc_t:
- if self.typename == str(exc).split('.')[-1]:
- return True
- return False
-
-class ReprOutcome(object):
- def __init__(self, repr_tuple):
- (self.passed, self.setupfailure, excinfo, self.skipped,
- self.is_critical, self.signal, self.stdout, self.stderr) = repr_tuple
- if excinfo is None:
- self.excinfo = None
- else:
- self.excinfo = ExcInfoRepr(excinfo)
-
- def __repr__(self):
- l = ["%s=%s" %(x, getattr(self, x))
- for x in "signal passed skipped setupfailure excinfo stdout stderr".split()]
- return "<ReprOutcome %s>" %(" ".join(l),)
Deleted: /py/trunk/py/test/rsession/repevent.py
==============================================================================
--- /py/trunk/py/test/rsession/repevent.py Mon Aug 27 11:02:50 2007
+++ (empty file)
@@ -1,145 +0,0 @@
-""" Reporter classes for showing asynchronous and synchronous status events
-"""
-
-import py
-import time
-
-def basic_report(msg_type, message):
- print msg_type, message
-
-#def report(msg_type, message):
-# pass
-
-##def report_error(excinfo):
-## if isinstance(excinfo, py.test.collect.Item.Skipped):
-## # we need to dispatch this info
-## report(Skipped(excinfo))
-## else:
-## report("itererror", excinfo)
-
-def wrapcall(reporter, func, *args, **kwargs):
- reporter(CallStart(func, args, kwargs))
- try:
- retval = func(*args, **kwargs)
- except:
- reporter(CallException(func, args, kwargs))
- raise
- else:
- reporter(CallFinish(func, args, kwargs))
- return retval
-
-# ----------------------------------------------------------------------
-# Reporting Events
-# ----------------------------------------------------------------------
-
-class ReportEvent(object):
- def __repr__(self):
- l = ["%s=%s" %(key, value)
- for key, value in self.__dict__.items()]
- return "<%s %s>" %(self.__class__.__name__, " ".join(l),)
-
-class SendItem(ReportEvent):
- def __init__(self, channel, item):
- self.item = item
- self.channel = channel
- if channel:
- self.host = channel.gateway.host
-
-class ReceivedItemOutcome(ReportEvent):
- def __init__(self, channel, item, outcome):
- self.channel = channel
- if channel:
- self.host = channel.gateway.host
- self.item = item
- self.outcome = outcome
-
-class CallEvent(ReportEvent):
- def __init__(self, func, args, kwargs):
- self.func = func
- self.args = args
- self.kwargs = kwargs
-
- def __repr__(self):
- call = "%s args=%s, kwargs=%s" %(self.func.__name__,
- self.args, self.kwargs)
- return '<%s %s>' %(self.__class__.__name__, call)
-
-class CallStart(CallEvent):
- pass
-
-class CallException(CallEvent):
- pass
-
-class CallFinish(CallEvent):
- pass
-
-class HostRSyncing(ReportEvent):
- def __init__(self, host, root, remotepath, synced):
- self.host = host
- self.root = root
- self.remotepath = remotepath
- self.synced = synced
-
-class HostGatewayReady(ReportEvent):
- def __init__(self, host, roots):
- self.host = host
- self.roots = roots
-
-class HostRSyncRootReady(ReportEvent):
- def __init__(self, host, root):
- self.host = host
- self.root = root
-
-class TestStarted(ReportEvent):
- def __init__(self, hosts, topdir, roots):
- self.hosts = hosts
- self.topdir = topdir
- self.roots = roots
- self.timestart = time.time()
-
-class TestFinished(ReportEvent):
- def __init__(self):
- self.timeend = time.time()
-
-class Nodes(ReportEvent):
- def __init__(self, nodes):
- self.nodes = nodes
-
-class SkippedTryiter(ReportEvent):
- def __init__(self, excinfo, item):
- self.excinfo = excinfo
- self.item = item
-
-class FailedTryiter(ReportEvent):
- def __init__(self, excinfo, item):
- self.excinfo = excinfo
- self.item = item
-
-class ItemStart(ReportEvent):
- """ This class shows most of the start stuff, like directory, module, class
- can be used for containers
- """
- def __init__(self, item):
- self.item = item
-
-class RsyncFinished(ReportEvent):
- def __init__(self):
- self.time = time.time()
-
-class ImmediateFailure(ReportEvent):
- def __init__(self, item, outcome):
- self.item = item
- self.outcome = outcome
-
-class PongReceived(ReportEvent):
- def __init__(self, hostid, result):
- self.hostid = hostid
- self.result = result
-
-class InterruptedExecution(ReportEvent):
- def __init__(self):
- self.timeend = time.time()
-
-class CrashedExecution(ReportEvent):
- def __init__(self):
- self.timeend = time.time()
Deleted: /py/trunk/py/test/rsession/reporter.py
==============================================================================
--- /py/trunk/py/test/rsession/reporter.py Mon Aug 27 11:02:50 2007
+++ (empty file)
@@ -1,347 +0,0 @@
-
-""" reporter - different reporter for different purposes ;-)
- Still lacks:
-
- 1. Hanging nodes are not done well
-"""
-
-import py
-
-from py.__.test.terminal.out import getout
-from py.__.test.rsession import repevent
-from py.__.test.rsession import outcome
-from py.__.misc.terminal_helper import ansi_print, get_terminal_width
-from py.__.test.representation import Presenter
-
-import sys
-import thread
-
-class AbstractReporter(object):
- def __init__(self, config, hosts):
- self.config = config
- self.hosts = hosts
- self.failed_tests_outcome = []
- self.skipped_tests_outcome = []
- self.out = getout(py.std.sys.stdout)
- self.presenter = Presenter(self.out, config)
- self.failed = dict([(host, 0) for host in hosts])
- self.skipped = dict([(host, 0) for host in hosts])
- self.passed = dict([(host, 0) for host in hosts])
- self.to_rsync = {}
-
- def get_item_name(self, event, colitem):
- return "/".join(colitem.listnames())
-
- def report(self, what):
- repfun = getattr(self, "report_" + what.__class__.__name__,
- self.report_unknown)
- try:
- return repfun(what)
- except (KeyboardInterrupt, SystemExit):
- raise
- except:
- print "Internal reporting problem"
- excinfo = py.code.ExceptionInfo()
- for i in excinfo.traceback:
- print str(i)[2:-1]
- print excinfo
-
- def report_unknown(self, what):
- if self.config.option.verbose:
- print "Unknown report: %s" % what
-
- def report_SendItem(self, item):
- address = item.host.hostname
- assert isinstance(item.host.hostname, str)
- if self.config.option.verbose:
- print "Sending %s to %s" % (item.item,
- address)
-
- def report_HostRSyncing(self, item):
- hostrepr = self._hostrepr(item.host)
- if item.synced:
- if (item.host.hostname == "localhost" and
- item.root == item.remotepath):
- print "%15s: skipping inplace rsync of %r" %(
- hostrepr, item.remotepath)
- else:
- print "%15s: skip duplicate rsync to %r" % (
- hostrepr, item.remotepath)
- else:
- print "%15s: rsync %r to remote %r" % (hostrepr,
- item.root.basename,
- item.remotepath)
-
- def report_HostGatewayReady(self, item):
- self.to_rsync[item.host] = len(item.roots)
- hostrepr = self._hostrepr(item.host)
- self.out.write("%15s: gateway initialised (remote topdir: %s)\n"\
- % (hostrepr, item.host.gw_remotepath))
-
- def report_HostRSyncRootReady(self, item):
- self.to_rsync[item.host] -= 1
- if not self.to_rsync[item.host]:
- self._host_ready(item)
-
- def _host_ready(self, item):
- self.hosts_to_rsync -= 1
- hostrepr = self._hostrepr(item.host)
- if self.hosts_to_rsync:
- print "%15s: READY (still %d to go)" % (hostrepr,
- self.hosts_to_rsync)
- else:
- print "%15s: READY" % hostrepr
-
- def report_TestStarted(self, item):
- hostreprs = [self._hostrepr(host) for host in item.hosts]
- txt = " Test started, hosts: %s " % ", ".join(hostreprs)
- self.hosts_to_rsync = len(item.hosts)
- self.out.sep("=", txt)
- self.timestart = item.timestart
- self.out.write("local top directory: %s\n" % item.topdir)
- for i, root in py.builtin.enumerate(item.roots):
- outof = "%d/%d" %(i+1, len(item.roots))
- self.out.write("local RSync root [%s]: %s\n" %
- (outof, root))
-
- 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()
- self.failures()
- if hasattr(self, 'nodes'): # XXX: Testing
- self.hangs()
- self.summary()
- return len(self.failed_tests_outcome) > 0
-
- report_InterruptedExecution = report_TestFinished
- report_CrashedExecution = report_TestFinished
-
- def hangs(self):
- h = []
- if self.config.option.exitfirst:
- # reporting hanging nodes in that case makes no sense at all
- # but we should share some code in all reporters than
- return
- for node in self.nodes:
- h += [(i, node.channel.gateway.sshaddress) for i in node.pending]
- if h:
- self.out.sep("=", " HANGING NODES ")
- for i, node in h:
- self.out.line("%s on %s" % (" ".join(i.listnames()), node))
-
- def failures(self):
- if self.failed_tests_outcome:
- self.out.sep("=", " FAILURES ")
- for event in self.failed_tests_outcome:
- if isinstance(event, repevent.ReceivedItemOutcome):
- host = self.gethost(event)
- self.out.sep('_', "%s on %s" %
- (" ".join(event.item.listnames()), host))
- if event.outcome.signal:
- self.presenter.repr_item_info(event.item)
- self.repr_signal(event.item, event.outcome)
- else:
- self.repr_failure(event.item, event.outcome)
- else:
- self.out.sep('_', " ".join(event.item.listnames()))
- out = outcome.Outcome(excinfo=event.excinfo)
- self.repr_failure(event.item, outcome.ReprOutcome(out.make_repr()))
-
- def gethost(self, event):
- return event.host.hostname
-
- def repr_failure(self, item, outcome):
- excinfo = outcome.excinfo
- traceback = excinfo.traceback
- if not traceback:
- self.out.line("empty traceback from item %r" % (item,))
- return
-
- handler = getattr(self.presenter, 'repr_failure_tb%s' % self.config.option.tbstyle)
- handler(item, excinfo, traceback, lambda: self.repr_out_err(outcome))
-
- def repr_out_err(self, outcome):
- if outcome.stdout:
- self.out.sep('-', " Captured process stdout: ")
- self.out.write(outcome.stdout)
- if outcome.stderr:
- self.out.sep('-', " Captured process stderr: ")
- self.out.write(outcome.stderr)
-
- def repr_signal(self, item, outcome):
- signal = outcome.signal
- self.out.line("Received signal: %d" % outcome.signal)
- self.repr_out_err(outcome)
-
- def _hostrepr(self, host):
- return host.hostid
-
- def skips(self):
- texts = {}
- for event in self.skipped_tests_outcome:
- colitem = event.item
- if isinstance(event, repevent.ReceivedItemOutcome):
- outcome = event.outcome
- text = outcome.skipped
- itemname = self.get_item_name(event, colitem)
- elif isinstance(event, repevent.SkippedTryiter):
- text = str(event.excinfo.value)
- itemname = "/".join(colitem.listnames())
- if text not in texts:
- texts[text] = [itemname]
- else:
- texts[text].append(itemname)
-
- if texts:
- self.out.line()
- self.out.sep('_', 'reasons for skipped tests')
- for text, items in texts.items():
- for item in items:
- self.out.line('Skipped in %s' % item)
- self.out.line("reason: %s" % text)
-
- def summary(self):
- def gather(dic):
- total = 0
- for key, val in dic.iteritems():
- total += val
- return total
-
- def create_str(name, count):
- if count:
- return ", %d %s" % (count, name)
- return ""
-
- total_passed = gather(self.passed)
- total_failed = gather(self.failed)
- total_skipped = gather(self.skipped)
- total = total_passed + total_failed + total_skipped
- skipped_str = create_str("skipped", total_skipped)
- failed_str = create_str("failed", total_failed)
- self.print_summary(total, skipped_str, failed_str)
-
- def print_summary(self, total, skipped_str, failed_str):
- self.out.sep("=", " %d test run%s%s in %.2fs (rsync: %.2f)" %
- (total, skipped_str, failed_str, self.timeend - self.timestart,
- self.timersync - self.timestart))
-
- def report_SkippedTryiter(self, event):
- #event.outcome.excinfo.source =
- self.skipped_tests_outcome.append(event)
-
- def report_FailedTryiter(self, event):
- pass
- # XXX: right now we do not do anything with it
-
- def report_ReceivedItemOutcome(self, event):
- host = event.host
- hostrepr = self._hostrepr(host)
- if event.outcome.passed:
- self.passed[host] += 1
- sys.stdout.write("%15s: PASSED " % hostrepr)
- elif event.outcome.skipped:
- self.skipped_tests_outcome.append(event)
- self.skipped[host] += 1
- sys.stdout.write("%15s: SKIPPED " % hostrepr)
- else:
- self.failed[host] += 1
- self.failed_tests_outcome.append(event)
- sys.stdout.write("%15s: " % hostrepr)
- ansi_print("FAILED", esc=(31,1), newline=False, file=sys.stdout)
- sys.stdout.write(" ")
- # we should have printed 20 characters to this point
- itempath = ".".join(event.item.listnames()[1:-1])
- funname = event.item.listnames()[-1]
- lgt = get_terminal_width() - 25
- # mark the function name, to be sure
- to_display = len(itempath) + len(funname) + 1
- if to_display > lgt:
- sys.stdout.write("..." + itempath[to_display-lgt+4:])
- else:
- sys.stdout.write(itempath)
- sys.stdout.write(" ")
- ansi_print(funname, esc=32, file=sys.stdout)
-
- def report_Nodes(self, event):
- self.nodes = event.nodes
-
-class RemoteReporter(AbstractReporter):
- def get_item_name(self, event, colitem):
- return event.host.hostname + ":" + \
- "/".join(colitem.listnames())
-
- def report_FailedTryiter(self, event):
- self.out.line("FAILED TO LOAD MODULE: %s\n" % "/".join(event.item.listnames()))
- self.failed_tests_outcome.append(event)
- # argh! bad hack, need to fix it
- self.failed[self.hosts[0]] += 1
-
- def report_SkippedTryiter(self, event):
- self.out.line("Skipped (%s) %s\n" % (str(event.excinfo.value), "/".
- join(event.item.listnames())))
-
-class LocalReporter(AbstractReporter):
- def get_item_name(self, event, colitem):
- return "/".join(colitem.listnames())
-
- def report_SkippedTryiter(self, event):
- #self.show_item(event.item, False)
- if isinstance(event.item, py.test.collect.Module):
- self.out.write("- skipped (%s)" % event.excinfo.value)
- else:
- self.out.write("s")
- self.skipped_tests_outcome.append(event)
-
- def report_FailedTryiter(self, event):
- #self.show_item(event.item, False)
- self.out.write("- FAILED TO LOAD MODULE")
- self.failed_tests_outcome.append(event)
- self.failed[self.hosts[0]] += 1
-
- def report_ReceivedItemOutcome(self, event):
- host = self.hosts[0]
- if event.outcome.passed:
- self.passed[host] += 1
- self.out.write(".")
- elif event.outcome.skipped:
- self.skipped_tests_outcome.append(event)
- self.skipped[host] += 1
- self.out.write("s")
- else:
- self.failed[host] += 1
- self.failed_tests_outcome.append(event)
- self.out.write("F")
-
- def report_ItemStart(self, event):
- self.show_item(event.item)
-
- def show_item(self, item, count_elems = True):
- if isinstance(item, py.test.collect.Module):
- # XXX This is a terrible hack, I don't like it
- # and will rewrite it at some point
- #self.count = 0
- lgt = len(list(item._tryiter()))
- #self.lgt = lgt
- # print names relative to current workdir
- name = "/".join(item.listnames())
- local = str(py.path.local())
- d = str(self.config.topdir)
- if local.startswith(d):
- local = local[len(d) + 1:]
- if local and name.startswith(local):
- name = name[len(local) + 1:]
- self.out.write("\n%s[%d] " % (name, lgt))
-
- def gethost(self, event):
- return 'localhost'
-
- def hangs(self):
- pass
Modified: py/trunk/py/test/rsession/rest.py
==============================================================================
--- py/trunk/py/test/rsession/rest.py (original)
+++ py/trunk/py/test/rsession/rest.py Mon Aug 27 11:02:50 2007
@@ -5,8 +5,8 @@
import py
import sys
from StringIO import StringIO
-from py.__.test.rsession.reporter import AbstractReporter
-from py.__.test.rsession import repevent
+from py.__.test.reporter import AbstractReporter
+from py.__.test import repevent
from py.__.rest.rst import *
class RestReporter(AbstractReporter):
Modified: py/trunk/py/test/rsession/rsession.py
==============================================================================
--- py/trunk/py/test/rsession/rsession.py (original)
+++ py/trunk/py/test/rsession/rsession.py Mon Aug 27 11:02:50 2007
@@ -8,101 +8,15 @@
import re
import time
-from py.__.test.rsession import repevent
+from py.__.test import repevent
from py.__.test.rsession.master import MasterNode, dispatch_loop, itemgen
from py.__.test.rsession.hostmanage import HostInfo, HostManager
from py.__.test.rsession.local import local_loop, plain_runner, apigen_runner,\
box_runner
-from py.__.test.rsession.reporter import LocalReporter, RemoteReporter
-from py.__.test.session import Session
+from py.__.test.reporter import LocalReporter, RemoteReporter, TestReporter
+from py.__.test.session import AbstractSession
from py.__.test.outcome import Skipped, Failed
-
-class AbstractSession(Session):
- """
- An abstract session executes collectors/items through a runner.
-
- """
- def fixoptions(self):
- option = self.config.option
- if option.runbrowser and not option.startserver:
- #print "--runbrowser implies --startserver"
- option.startserver = True
- if self.config.getvalue("dist_boxed"):
- option.boxed = True
- super(AbstractSession, self).fixoptions()
-
- def init_reporter(self, reporter, hosts, reporter_class, arg=""):
- """ This initialises so called `reporter` class, which will
- handle all event presenting to user. Does not get called
- if main received custom reporter
- """
- startserverflag = self.config.option.startserver
- restflag = self.config.option.restreport
-
- if startserverflag and reporter is None:
- from py.__.test.rsession.web import start_server, exported_methods
- if self.config.option.runbrowser:
- from socket import INADDR_ANY
- port = INADDR_ANY # pick a random port when starting browser
- else:
- port = 8000 # stick to a fixed port otherwise
-
- reporter = exported_methods.report
- httpd = start_server(server_address = ('', port))
- port = httpd.server_port
- if self.config.option.runbrowser:
- import webbrowser, thread
- # webbrowser.open() may block until the browser finishes or not
- url = "http://localhost:%d" % (port,)
- thread.start_new_thread(webbrowser.open, (url,))
- elif reporter is None:
- if restflag:
- from py.__.test.rsession.rest import RestReporter
- reporter_class = RestReporter
- if arg:
- reporter_instance = reporter_class(self.config, hosts)
- else:
- reporter_instance = reporter_class(self.config, hosts)
- reporter = reporter_instance.report
- else:
- startserverflag = False
-
- return reporter, startserverflag
- def reporterror(reporter, data):
- excinfo, item = data
- if excinfo is None:
- reporter(repevent.ItemStart(item))
- elif excinfo.type is Skipped:
- reporter(repevent.SkippedTryiter(excinfo, item))
- else:
- reporter(repevent.FailedTryiter(excinfo, item))
- reporterror = staticmethod(reporterror)
-
- def kill_server(self, startserverflag):
- """ Kill web server
- """
- if startserverflag:
- from py.__.test.rsession.web import kill_server
- kill_server()
-
- def wrap_reporter(self, reporter):
- """ We wrap reporter around, which makes it possible to us to track
- existance of failures
- """
- self.was_failure = False
- def new_reporter(event):
- if isinstance(event, repevent.ReceivedItemOutcome) and \
- not event.outcome.passed and \
- not event.outcome.skipped:
- self.was_failure = True
- return reporter(event)
- checkfun = lambda : self.config.option.exitfirst and \
- self.was_failure
- # for tests
- self.checkfun = checkfun
- return new_reporter, checkfun
-
class RSession(AbstractSession):
""" Remote version of session
"""
@@ -129,12 +43,9 @@
def main(self, reporter=None):
""" main loop for running tests. """
- args = self.config.args
-
- hm = HostManager(self.config)
- reporter, startserverflag = self.init_reporter(reporter,
- hm.hosts, RemoteReporter)
- reporter, checkfun = self.wrap_reporter(reporter)
+ config = self.config
+ hm = HostManager(config)
+ reporter, checkfun = self.init_reporter(reporter, config, hm.hosts)
reporter(repevent.TestStarted(hm.hosts, self.config.topdir,
hm.roots))
@@ -157,22 +68,18 @@
exitfirst=self.config.option.exitfirst)
reporter(repevent.Nodes(nodes))
retval = reporter(repevent.TestFinished())
- self.kill_server(startserverflag)
return retval
except (KeyboardInterrupt, SystemExit):
reporter(repevent.InterruptedExecution())
- self.kill_server(startserverflag)
raise
except:
reporter(repevent.CrashedExecution())
- self.kill_server(startserverflag)
raise
def dispatch_tests(self, nodes, reporter, checkfun):
colitems = self.config.getcolitems()
keyword = self.config.option.keyword
- itemgenerator = itemgen(colitems, reporter, keyword, self.reporterror)
-
+ itemgenerator = itemgen(colitems, reporter, keyword)
all_tests = dispatch_loop(nodes, itemgenerator, checkfun)
class LSession(AbstractSession):
@@ -180,16 +87,13 @@
"""
def main(self, reporter=None, runner=None):
# check out if used options makes any sense
- args = self.config.args
-
- hm = HostManager(self.config, hosts=[HostInfo('localhost')])
+ config = self.config
+ hm = HostManager(config, hosts=[HostInfo('localhost')])
hosts = hm.hosts
if not self.config.option.nomagic:
py.magic.invoke(assertion=1)
- reporter, startserverflag = self.init_reporter(reporter,
- hosts, LocalReporter, args[0])
- reporter, checkfun = self.wrap_reporter(reporter)
+ reporter, checkfun = self.init_reporter(reporter, config, hosts)
reporter(repevent.TestStarted(hosts, self.config.topdir, []))
colitems = self.config.getcolitems()
@@ -200,11 +104,10 @@
keyword = self.config.option.keyword
- itemgenerator = itemgen(colitems, reporter, keyword, self.reporterror)
+ itemgenerator = itemgen(colitems, reporter, keyword)
local_loop(self, reporter, itemgenerator, checkfun, self.config, runner=runner)
retval = reporter(repevent.TestFinished())
- self.kill_server(startserverflag)
if not self.config.option.nomagic:
py.magic.revoke(assertion=1)
Modified: py/trunk/py/test/rsession/slave.py
==============================================================================
--- py/trunk/py/test/rsession/slave.py (original)
+++ py/trunk/py/test/rsession/slave.py Mon Aug 27 11:02:50 2007
@@ -4,7 +4,7 @@
import py
from py.__.test.rsession.executor import RunExecutor, BoxExecutor, AsyncExecutor
-from py.__.test.rsession.outcome import Outcome
+from py.__.test.outcome import SerializableOutcome
from py.__.test.outcome import Skipped
import thread
import os
@@ -53,10 +53,10 @@
node = getnode(nextitem)
res = node.run(nextitem)
except Skipped, s:
- send(Outcome(skipped=str(s)).make_repr())
+ send(SerializableOutcome(skipped=str(s)).make_repr())
except:
excinfo = py.code.ExceptionInfo()
- send(Outcome(excinfo=excinfo, is_critical=True).make_repr())
+ send(SerializableOutcome(excinfo=excinfo, is_critical=True).make_repr())
else:
if not res[0] and not res[3] and config.option.exitfirst:
# we're finished, but need to eat what we can
Modified: py/trunk/py/test/rsession/testing/test_executor.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_executor.py (original)
+++ py/trunk/py/test/rsession/testing/test_executor.py Mon Aug 27 11:02:50 2007
@@ -4,7 +4,7 @@
from py.__.test.rsession.executor import RunExecutor, BoxExecutor,\
AsyncExecutor, ApigenExecutor
-from py.__.test.rsession.outcome import ReprOutcome
+from py.__.test.outcome import ReprOutcome
from py.__.test.rsession.testing.basetest import BasicRsessionTest
from py.__.test.outcome import Failed
Modified: py/trunk/py/test/rsession/testing/test_hostmanage.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_hostmanage.py (original)
+++ py/trunk/py/test/rsession/testing/test_hostmanage.py Mon Aug 27 11:02:50 2007
@@ -5,7 +5,7 @@
import py
from py.__.test.rsession.hostmanage import HostRSync, HostInfo, HostManager
from py.__.test.rsession.hostmanage import sethomedir, gethomedir, getpath_relto_home
-from py.__.test.rsession import repevent
+from py.__.test import repevent
class DirSetup:
def setup_method(self, method):
Modified: py/trunk/py/test/rsession/testing/test_lsession.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_lsession.py (original)
+++ py/trunk/py/test/rsession/testing/test_lsession.py Mon Aug 27 11:02:50 2007
@@ -4,7 +4,7 @@
import py
from py.__.test.rsession.rsession import LSession
-from py.__.test.rsession import repevent
+from py.__.test import repevent
from py.__.test.rsession.local import box_runner, plain_runner, apigen_runner
def setup_module(mod):
Modified: py/trunk/py/test/rsession/testing/test_master.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_master.py (original)
+++ py/trunk/py/test/rsession/testing/test_master.py Mon Aug 27 11:02:50 2007
@@ -11,8 +11,8 @@
from py.__.test.rsession.master import dispatch_loop, MasterNode
from py.__.test.rsession.slave import setup_slave
-from py.__.test.rsession.outcome import ReprOutcome, Outcome
-from py.__.test.rsession import repevent
+from py.__.test.outcome import ReprOutcome, SerializableOutcome
+from py.__.test import repevent
from py.__.test.rsession.hostmanage import HostInfo
def setup_module(mod):
@@ -64,8 +64,8 @@
mnode = MasterNode(ch, reportlist.append)
mnode.send(Item("ok"))
mnode.send(Item("notok"))
- ch.callback(Outcome().make_repr())
- ch.callback(Outcome(excinfo=excinfo).make_repr())
+ ch.callback(SerializableOutcome().make_repr())
+ ch.callback(SerializableOutcome(excinfo=excinfo).make_repr())
assert len(reportlist) == 4
received = [i for i in reportlist
if isinstance(i, repevent.ReceivedItemOutcome)]
@@ -91,12 +91,12 @@
mnode = MasterNode(ch, reportlist.append)
mnode.send(Item("ok"))
mnode.send(Item("ok"))
- ch.callback(Outcome().make_repr())
- ch.callback(Outcome().make_repr())
+ ch.callback(SerializableOutcome().make_repr())
+ ch.callback(SerializableOutcome().make_repr())
assert len(reportlist) == 4
def test_outcome_repr():
- out = ReprOutcome(Outcome(skipped=True).make_repr())
+ out = ReprOutcome(SerializableOutcome(skipped=True).make_repr())
s = repr(out)
assert s.lower().find("skip") != -1
Deleted: /py/trunk/py/test/rsession/testing/test_outcome.py
==============================================================================
--- /py/trunk/py/test/rsession/testing/test_outcome.py Mon Aug 27 11:02:50 2007
+++ (empty file)
@@ -1,50 +0,0 @@
-
-import py
-from py.__.test.rsession.outcome import Outcome, ReprOutcome, ExcInfoRepr
-
-import marshal
-
-def test_critical_debugging_flag():
- outcome = Outcome(is_critical=True)
- r = ReprOutcome(outcome.make_repr())
- assert r.is_critical
-
-def f1():
- 1
- 2
- 3
- 4
- raise ValueError(42)
-
-def f2():
- f1()
-
-def f3():
- f2()
-
-def test_exception_info_repr():
- try:
- f3()
- except:
- outcome = Outcome(excinfo=py.code.ExceptionInfo())
-
- repr = outcome.make_excinfo_repr("long")
- assert marshal.dumps(repr)
- excinfo = ExcInfoRepr(repr)
-
- assert str(excinfo.typename) == "ValueError"
- assert str(excinfo.value) == "42"
- assert len(excinfo.traceback) == 4
- myfile = py.magic.autopath()
- assert excinfo.traceback[3].path == myfile
- assert excinfo.traceback[3].lineno == f1.func_code.co_firstlineno + 4
- assert excinfo.traceback[3].relline == 5
- assert excinfo.traceback[2].path == myfile
- assert excinfo.traceback[2].lineno == f2.func_code.co_firstlineno
- assert excinfo.traceback[2].relline == 1
- assert excinfo.traceback[1].path == myfile
- assert excinfo.traceback[1].lineno == f3.func_code.co_firstlineno
- assert excinfo.traceback[1].relline == 1
-
-#def test_f3():
-# f3()
Deleted: /py/trunk/py/test/rsession/testing/test_repevent.py
==============================================================================
--- /py/trunk/py/test/rsession/testing/test_repevent.py Mon Aug 27 11:02:50 2007
+++ (empty file)
@@ -1,36 +0,0 @@
-""" test reporting functionality. """
-
-import py
-from py.__.test.rsession import repevent
-
-def test_wrapcall_ok():
- l = []
- def ok(x):
- return x+1
- i = repevent.wrapcall(l.append, ok, 1)
- assert i == 2
- assert len(l) == 2
- assert isinstance(l[0], repevent.CallStart)
- assert isinstance(l[1], repevent.CallFinish)
- assert repr(l[0])
- assert repr(l[1])
-
-def test_wrapcall_exception():
- l = []
- def fail(x):
- raise ValueError
- py.test.raises(ValueError, "repevent.wrapcall(l.append, fail, 1)")
- assert len(l) == 2
- assert isinstance(l[0], repevent.CallStart)
- assert isinstance(l[1], repevent.CallException)
-
-def test_reporter_methods_sanity():
- """ Checks if all the methods of reporter are sane
- """
- from py.__.test.rsession.rsession import RemoteReporter
- from py.__.test.rsession import repevent
-
- for method in dir(RemoteReporter):
-
- if method.startswith("report_") and method != "report_unknown":
- assert method[len('report_'):] in repevent.__dict__
Deleted: /py/trunk/py/test/rsession/testing/test_reporter.py
==============================================================================
--- /py/trunk/py/test/rsession/testing/test_reporter.py Mon Aug 27 11:02:50 2007
+++ (empty file)
@@ -1,216 +0,0 @@
-
-""" reporter tests.
-
-XXX there are a few disabled reporting tests because
-they test for exact formatting as far as i can see.
-I think it's rather better to directly invoke a
-reporter and pass it some hand-prepared events to see
-that running the reporter doesn't break shallowly.
-
-Otherwise, i suppose that some "visual" testing can usually be driven
-manually by user-input. And when passing particular events
-to a reporter it's also easier to check for one line
-instead of having to know the order in which things are printed
-etc.
-
-
-"""
-
-
-import py, os
-from py.__.test.rsession.rsession import LocalReporter, AbstractSession,\
- RemoteReporter
-from py.__.test.rsession import repevent
-from py.__.test.rsession.outcome import ReprOutcome, Outcome
-from py.__.test.rsession.hostmanage import HostInfo
-from py.__.test.rsession.box import Box
-from py.__.test.rsession.testing.basetest import BasicRsessionTest
-import sys
-from StringIO import StringIO
-
-class DummyGateway(object):
- def __init__(self, host):
- self.host = host
-
-class DummyChannel(object):
- def __init__(self, host):
- self.gateway = DummyGateway(host)
-
-class AbstractTestReporter(BasicRsessionTest):
- def prepare_outcomes(self):
- # possible outcomes
- try:
- 1/0
- except:
- exc = py.code.ExceptionInfo()
-
- outcomes = [Outcome(()),
- Outcome(skipped=True),
- Outcome(excinfo=exc),
- Outcome()]
-
- outcomes = [ReprOutcome(outcome.make_repr()) for outcome in outcomes]
- outcomes[3].signal = 11
- outcomes[0].passed = False
-
- return outcomes
-
- def report_received_item_outcome(self):
- item = self.getexample("pass")
- outcomes = self.prepare_outcomes()
-
- def boxfun(config, item, outcomes):
- hosts = [HostInfo("localhost")]
- r = self.reporter(config, hosts)
- ch = DummyChannel(hosts[0])
- for outcome in outcomes:
- r.report(repevent.ReceivedItemOutcome(ch, item, outcome))
-
- cap = py.io.StdCaptureFD()
- boxfun(self.config, item, outcomes)
- out, err = cap.reset()
- assert not err
- return out
-
- def _test_module(self):
- funcitem = self.getexample("pass")
- moditem = self.getmod()
- outcomes = self.prepare_outcomes()
-
- def boxfun(config, item, funcitem, outcomes):
- hosts = [HostInfo('localhost')]
- r = self.reporter(config, hosts)
- r.report(repevent.ItemStart(item))
- ch = DummyChannel(hosts[0])
- for outcome in outcomes:
- r.report(repevent.ReceivedItemOutcome(ch, funcitem, outcome))
-
- cap = py.io.StdCaptureFD()
- boxfun(self.config, moditem, funcitem, outcomes)
- out, err = cap.reset()
- assert not err
- return out
-
- def _test_full_module(self):
- tmpdir = py.test.ensuretemp("repmod")
- tmpdir.ensure("__init__.py")
- tmpdir.ensure("test_one.py").write(py.code.Source("""
- def test_x():
- pass
- """))
- tmpdir.ensure("test_two.py").write(py.code.Source("""
- import py
- py.test.skip("reason")
- """))
- tmpdir.ensure("test_three.py").write(py.code.Source("""
- sadsadsa
- """))
-
- def boxfun():
- config = py.test.config._reparse([str(tmpdir)])
- rootcol = py.test.collect.Directory(tmpdir)
- hosts = [HostInfo('localhost')]
- r = self.reporter(config, hosts)
- list(rootcol._tryiter(reporterror=lambda x : AbstractSession.reporterror(r.report, x)))
-
- cap = py.io.StdCaptureFD()
- boxfun()
- out, err = cap.reset()
- assert not err
- return out
-
- def test_failed_to_load(self):
- tmpdir = py.test.ensuretemp("failedtoload")
- tmpdir.ensure("__init__.py")
- tmpdir.ensure("test_three.py").write(py.code.Source("""
- sadsadsa
- """))
- def boxfun():
- config = py.test.config._reparse([str(tmpdir)])
- rootcol = py.test.collect.Directory(tmpdir)
- host = HostInfo('localhost')
- r = self.reporter(config, [host])
- r.report(repevent.TestStarted([host], config.topdir, ["a"]))
- r.report(repevent.RsyncFinished())
- list(rootcol._tryiter(reporterror=lambda x : AbstractSession.reporterror(r.report, x)))
- r.report(repevent.TestFinished())
- return r
-
- cap = py.io.StdCaptureFD()
- r = boxfun()
- out, err = cap.reset()
- assert not err
- assert out.find("1 failed in") != -1
- assert out.find("NameError: name 'sadsadsa' is not defined") != -1
-
- def _test_still_to_go(self):
- tmpdir = py.test.ensuretemp("stilltogo")
- tmpdir.ensure("__init__.py")
- cap = py.io.StdCaptureFD()
- config = py.test.config._reparse([str(tmpdir)])
- hosts = [HostInfo(i) for i in ["host1", "host2", "host3"]]
- r = self.reporter(config, hosts)
- r.report(repevent.TestStarted(hosts, config.topdir, ["a", "b", "c"]))
- for host in hosts:
- r.report(repevent.HostGatewayReady(host, ["a", "b", "c"]))
- for host in hosts:
- for root in ["a", "b", "c"]:
- r.report(repevent.HostRSyncRootReady(host, root))
- out, err = cap.reset()
- assert not err
- expected1 = "Test started, hosts: host1[0], host2[0], host3[0]"
- assert out.find(expected1) != -1
- for expected in py.code.Source("""
- host1[0]: READY (still 2 to go)
- host2[0]: READY (still 1 to go)
- host3[0]: READY
- """).lines:
- expected = expected.strip()
- assert out.find(expected) != -1
-
-class TestLocalReporter(AbstractTestReporter):
- reporter = LocalReporter
-
- def test_report_received_item_outcome(self):
- assert self.report_received_item_outcome() == 'FsF.'
-
- def test_module(self):
- output = self._test_module()
- assert output.find("test_one") != -1
- assert output.endswith("FsF."), output
-
- def test_full_module(self):
- received = self._test_full_module()
- expected_lst = ["repmod/test_one.py", "FAILED TO LOAD MODULE",
- "skipped", "reason"]
- for i in expected_lst:
- assert received.find(i) != -1
-
-class TestRemoteReporter(AbstractTestReporter):
- reporter = RemoteReporter
-
- def test_still_to_go(self):
- self._test_still_to_go()
-
- def test_report_received_item_outcome(self):
- val = self.report_received_item_outcome()
- expected_lst = ["localhost", "FAILED",
- "funcpass", "test_one",
- "SKIPPED",
- "PASSED"]
- for expected in expected_lst:
- assert val.find(expected) != -1
-
- def test_module(self):
- val = self._test_module()
- expected_lst = ["localhost", "FAILED",
- "funcpass", "test_one",
- "SKIPPED",
- "PASSED"]
- for expected in expected_lst:
- assert val.find(expected) != -1
-
- def test_full_module(self):
- val = self._test_full_module()
- assert val.find("FAILED TO LOAD MODULE: repmod/test_three.py\n"\
- "\nSkipped ('reason') repmod/test_two.py") != -1
Modified: py/trunk/py/test/rsession/testing/test_rest.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_rest.py (original)
+++ py/trunk/py/test/rsession/testing/test_rest.py Mon Aug 27 11:02:50 2007
@@ -3,13 +3,13 @@
"""
import py
-from py.__.test.rsession.testing.test_reporter import AbstractTestReporter,\
+from py.__.test.testing.test_reporter import AbstractTestReporter,\
DummyChannel
-from py.__.test.rsession import repevent
+from py.__.test import repevent
from py.__.test.rsession.rest import RestReporter, NoLinkWriter
from py.__.rest.rst import *
from py.__.test.rsession.hostmanage import HostInfo
-from py.__.test.rsession.outcome import Outcome
+from py.__.test.outcome import SerializableOutcome
class Container(object):
def __init__(self, **args):
@@ -109,7 +109,7 @@
"""
def test_ReceivedItemOutcome_PASSED(self):
- outcome = Outcome()
+ outcome = SerializableOutcome()
item = Container(listnames=lambda: ['', 'foo.py', 'bar', '()', 'baz'])
event = repevent.ReceivedItemOutcome(channel=ch, outcome=outcome, item=item)
reporter.report(event)
@@ -117,7 +117,7 @@
'foo.py/bar()/baz\n\n')
def test_ReceivedItemOutcome_SKIPPED(self):
- outcome = Outcome(skipped="reason")
+ outcome = SerializableOutcome(skipped="reason")
item = Container(listnames=lambda: ['', 'foo.py', 'bar', '()', 'baz'])
event = repevent.ReceivedItemOutcome(channel=ch, outcome=outcome, item=item)
reporter.report(event)
@@ -125,7 +125,7 @@
'foo.py/bar()/baz\n\n')
def test_ReceivedItemOutcome_FAILED(self):
- outcome = Outcome(excinfo="xxx")
+ outcome = SerializableOutcome(excinfo="xxx")
item = Container(listnames=lambda: ['', 'foo.py', 'bar', '()', 'baz'])
event = repevent.ReceivedItemOutcome(channel=ch, outcome=outcome, item=item)
reporter.report(event)
@@ -153,7 +153,7 @@
),
]
)
- outcome = Outcome(excinfo=excinfo)
+ outcome = SerializableOutcome(excinfo=excinfo)
outcome.stdout = '<printed>'
outcome.stderr = ''
parent = Container(parent=None, fspath=py.path.local('.'))
@@ -336,18 +336,15 @@
py.test.skip("Not implemented")
def test_report_received_item_outcome(self):
- py.test.skip("Relying on exact output matching")
val = self.report_received_item_outcome()
- expected = """\
-* localhost\: **FAILED** `traceback0`_\n py/test/rsession/testing/test\_slave.py/funcpass
-
-* localhost\: **SKIPPED** py/test/rsession/testing/test\_slave.py/funcpass
-
-* localhost\: **FAILED** `traceback1`_\n py/test/rsession/testing/test\_slave.py/funcpass
+ expected_list = [
+ "**FAILED**",
+ "**SKIPPED**",
+ "**PASSED**",
+ "* localhost\:",
+ "`traceback0`_ test\_one.py/funcpass",
+ "test\_one.py/funcpass"]
+ for expected in expected_list:
+ assert val.find(expected) != -1
-* localhost\: **PASSED** py/test/rsession/testing/test\_slave.py/funcpass
-
-"""
- print val
- assert val == expected
Modified: py/trunk/py/test/rsession/testing/test_rsession.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_rsession.py (original)
+++ py/trunk/py/test/rsession/testing/test_rsession.py Mon Aug 27 11:02:50 2007
@@ -3,7 +3,7 @@
"""
import py
-from py.__.test.rsession import repevent
+from py.__.test import repevent
from py.__.test.rsession.rsession import RSession
from py.__.test.rsession.hostmanage import HostManager, HostInfo
from py.__.test.rsession.testing.basetest import BasicRsessionTest
@@ -14,23 +14,6 @@
if py.std.sys.platform == "win32":
py.test.skip("rsession tests disabled for win32")
-def test_example_tryiter():
- events = []
- tmpdir = py.test.ensuretemp("tryitertest")
- tmpdir.ensure("a", "__init__.py")
- tmpdir.ensure("conftest.py").write(py.code.Source("""
- import py
- py.test.skip("Reason")
- """))
- tmpdir.ensure("a", "test_empty.py").write(py.code.Source("""
- def test_empty():
- pass
- """))
- rootcol = py.test.collect.Directory(tmpdir)
- data = list(rootcol._tryiter(reporterror=events.append))
- assert len(events) == 2
- assert str(events[1][0].value).find("Reason") != -1
-
class TestRSessionRemote(DirSetup, BasicRsessionTest):
def test_example_distribution_minus_x(self):
self.source.ensure("sub", "conftest.py").write(py.code.Source("""
@@ -57,7 +40,6 @@
testevents = [x for x in allevents
if isinstance(x, repevent.ReceivedItemOutcome)]
assert len(testevents) == 3
- assert rsession.checkfun()
def test_distribution_rsync_roots_example(self):
destdir = py.test.ensuretemp("example_dist_destdir")
Modified: py/trunk/py/test/rsession/testing/test_slave.py
==============================================================================
--- py/trunk/py/test/rsession/testing/test_slave.py (original)
+++ py/trunk/py/test/rsession/testing/test_slave.py Mon Aug 27 11:02:50 2007
@@ -1,7 +1,7 @@
""" Testing the slave side node code (in a local way). """
from py.__.test.rsession.slave import SlaveNode, slave_main, setup
-from py.__.test.rsession.outcome import ReprOutcome
+from py.__.test.outcome import ReprOutcome
import py, sys
from py.__.test.rsession.testing.basetest import BasicRsessionTest
Modified: py/trunk/py/test/rsession/web.py
==============================================================================
--- py/trunk/py/test/rsession/web.py (original)
+++ py/trunk/py/test/rsession/web.py Mon Aug 27 11:02:50 2007
@@ -15,7 +15,7 @@
import py
from py.__.test.rsession.rsession import RSession
-from py.__.test.rsession import repevent
+from py.__.test import repevent
from py.__.test import collect
from py.__.test.rsession.webdata import json
@@ -305,9 +305,8 @@
if not self.to_rsync[item.host]:
self._host_ready(item)
-
def report_TestStarted(self, event):
- # XXX: It overrides out self.hosts
+ # XXX: It overrides our self.hosts
self.hosts = {}
self.ready_hosts = {}
for host in event.hosts:
@@ -315,6 +314,13 @@
self.ready_hosts[host] = False
self.start_event.set()
self.pending_events.put(event)
+
+ def report_TestFinished(self, event):
+ self.pending_events.put(event)
+ kill_server()
+
+ report_InterruptedExecution = report_TestFinished
+ report_CrashedExecution = report_TestFinished
def report(self, what):
repfun = getattr(self, "report_" + what.__class__.__name__,
@@ -330,13 +336,6 @@
print str(i)[2:-1]
print excinfo
-## try:
-## self.wait_flag.acquire()
-## self.pending_events.insert(0, event)
-## self.wait_flag.notify()
-## finally:
-## self.wait_flag.release()
-
exported_methods = ExportedMethods()
class TestHandler(BaseHTTPRequestHandler):
@@ -400,7 +399,7 @@
js_name = py.path.local(__file__).dirpath("webdata").join("source.js")
web_name = py.path.local(__file__).dirpath().join("webjs.py")
if IMPORTED_PYPY and web_name.mtime() > js_name.mtime() or \
- (not js_name.check()) or 1:
+ (not js_name.check()):
from py.__.test.rsession import webjs
javascript_source = rpython2javascript(webjs,
@@ -418,6 +417,34 @@
self.end_headers()
self.wfile.write(data)
+class WebReporter(object):
+ """ A simple wrapper, this file needs ton of refactoring
+ anyway, so this is just to satisfy things below
+ (and start to create saner interface as well)
+ """
+ def __init__(self, config, hosts):
+ start_server_from_config(config)
+
+ # rebind
+ report = exported_methods.report
+ __call__ = report
+
+def start_server_from_config(config):
+ if config.option.runbrowser:
+ port = socket.INADDR_ANY
+ else:
+ port = 8000
+
+ httpd = start_server(server_address = ('', port))
+ port = httpd.server_port
+ if config.option.runbrowser:
+ import webbrowser, thread
+ # webbrowser.open() may block until the browser finishes or not
+ url = "http://localhost:%d" % (port,)
+ thread.start_new_thread(webbrowser.open, (url,))
+
+ return exported_methods.report
+
def start_server(server_address = ('', 8000), handler=TestHandler, start_new=True):
httpd = HTTPServer(server_address, handler)
Modified: py/trunk/py/test/rsession/webdata/source.js
==============================================================================
Binary files. No diff available.
Modified: py/trunk/py/test/session.py
==============================================================================
--- py/trunk/py/test/session.py (original)
+++ py/trunk/py/test/session.py Mon Aug 27 11:02:50 2007
@@ -1,33 +1,23 @@
import py
from py.__.test.outcome import Outcome, Failed, Passed, Skipped
+from py.__.test.reporter import choose_reporter, TestReporter
-class Session(object):
- """
- A Session gets test Items from Collectors, # executes the
- Items and sends the Outcome to the Reporter.
+class AbstractSession(object):
+ """ An abstract session executes collectors/items through a runner.
"""
def __init__(self, config):
self._memo = []
self.config = config
self._keyword = config.option.keyword
- def shouldclose(self):
- return False
-
- def header(self, colitems):
- """ setup any neccessary resources ahead of the test run. """
- if not self.config.option.nomagic:
- py.magic.invoke(assertion=1)
-
- def footer(self, colitems):
- """ teardown any resources after a test run. """
- py.test.collect.Function._state.teardown_all()
- if not self.config.option.nomagic:
- py.magic.revoke(assertion=1)
-
def fixoptions(self):
""" check, fix and determine conflicting options. """
- option = self.config.option
+ option = self.config.option
+ if option.runbrowser and not option.startserver:
+ #print "--runbrowser implies --startserver"
+ option.startserver = True
+ if self.config.getvalue("dist_boxed") and option.dist:
+ option.boxed = True
# implied options
if option.usepdb:
if not option.nocapture:
@@ -43,6 +33,34 @@
if option.keyword_oneshot and not option.keyword:
raise ValueError, "--keyword-oneshot makes sense only when --keyword is supplied"
+ def init_reporter(self, reporter, config, hosts):
+ if reporter is None:
+ reporter = choose_reporter(config)(config, hosts)
+ else:
+ reporter = TestReporter(reporter)
+ checkfun = lambda : self.config.option.exitfirst and \
+ reporter.was_failure()
+ return reporter, checkfun
+
+class Session(AbstractSession):
+ """
+ A Session gets test Items from Collectors, # executes the
+ Items and sends the Outcome to the Reporter.
+ """
+ def shouldclose(self):
+ return False
+
+ def header(self, colitems):
+ """ setup any neccessary resources ahead of the test run. """
+ if not self.config.option.nomagic:
+ py.magic.invoke(assertion=1)
+
+ def footer(self, colitems):
+ """ teardown any resources after a test run. """
+ py.test.collect.Function._state.teardown_all()
+ if not self.config.option.nomagic:
+ py.magic.revoke(assertion=1)
+
def start(self, colitem):
""" hook invoked before each colitem.run() invocation. """
Modified: py/trunk/py/test/testing/test_collect.py
==============================================================================
--- py/trunk/py/test/testing/test_collect.py (original)
+++ py/trunk/py/test/testing/test_collect.py Mon Aug 27 11:02:50 2007
@@ -7,10 +7,6 @@
mod.datadir = setupdatadir()
mod.tmpdir = py.test.ensuretemp('test_collect')
-def skipboxed():
- if py.test.config.option.boxed:
- py.test.skip("test does not work with boxed tests")
-
def test_failing_import_execfile():
dest = datadir / 'failingimport.py'
col = py.test.collect.Module(dest)
@@ -375,10 +371,6 @@
py.test.fail("should not have raised: %s" %(exc,))
l = []
- list(col._tryiter(reporterror=l.append))
- assert len(l) == 2
- excinfo, item = l[-1]
- assert isinstance(excinfo, py.code.ExceptionInfo)
def test_tryiter_handles_keyboardinterrupt():
tmp = py.test.ensuretemp("tryiterkeyboard")
@@ -416,9 +408,25 @@
"""))
tmp.ensure("__init__.py")
col = py.test.collect.Module(tmp.join("test_one.py"))
- errors = []
- l = list(col._tryiter(reporterror=errors.append))
- assert len(errors) == 2
+ assert len(col.join('test_one').run()) == 3
+
+def test_generator_setup_invoked_twice():
+ py.test.skip("Test for generators not invoking setup, needs thinking")
+ tmp = py.test.ensuretemp("generator_setup_invoke")
+ tmp.ensure("test_one.py").write(py.code.Source("""
+ def setup_module(mod):
+ mod.x = []
+
+ def setup_function(fun):
+ x.append(1)
+
+ def test_one():
+ yield lambda: None
+ """))
+ tmp.ensure("__init__.py")
+ col = py.test.collect.Module(tmp.join("test_one.py"))
+ l = list(col._tryiter())
+ assert not hasattr(col.obj, 'x')
def test_check_collect_hashes():
tmp = py.test.ensuretemp("check_collect_hashes")
Modified: py/trunk/py/test/testing/test_config.py
==============================================================================
--- py/trunk/py/test/testing/test_config.py (original)
+++ py/trunk/py/test/testing/test_config.py Mon Aug 27 11:02:50 2007
@@ -2,7 +2,6 @@
import py
from py.__.test.config import gettopdir
-from py.__.test.testing.test_collect import skipboxed
def test_tmpdir():
d1 = py.test.ensuretemp('hello')
More information about the pytest-commit
mailing list