floraachy
/
pytest2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[svn r46011] 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

--HG--
branch : trunk
This commit is contained in:
fijal 2007-08-27 11:02:50 +02:00
parent 46fdbe7867
commit 28c5aae67d
27 changed files with 4266 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
py/test/collect.py
View File

@ -189,7 +189,7 @@ class Collector(object):
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 @@ class Collector(object):
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))
pass
def _getsortvalue(self):
return self.name

122
py/test/outcome.py
View File

@ -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 ExceptionFailure(Failed):
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),)

9
py/test/rsession/repevent.py → py/test/repevent.py
View File

@ -38,6 +38,9 @@ class ReportEvent(object):
for key, value in self.__dict__.items()]
return "<%s %s>" %(self.__class__.__name__, " ".join(l),)
def is_failure(self):
return False
class SendItem(ReportEvent):
def __init__(self, channel, item):
self.item = item
@ -53,6 +56,9 @@ class ReceivedItemOutcome(ReportEvent):
self.item = item
self.outcome = outcome
def is_failure(self):
return not (self.outcome.passed or self.outcome.skipped)
class CallEvent(ReportEvent):
def __init__(self, func, args, kwargs):
self.func = func
@ -115,6 +121,9 @@ class FailedTryiter(ReportEvent):
self.excinfo = excinfo
self.item = item
def is_failure(self):
return True
class ItemStart(ReportEvent):
""" This class shows most of the start stuff, like directory, module, class
can be used for containers

50
py/test/rsession/reporter.py → py/test/reporter.py
View File

@ -8,13 +8,45 @@
import py
from py.__.test.terminal.out import getout
from py.__.test.rsession import repevent
from py.__.test.rsession import outcome
from py.__.test import repevent
from py.__.test import outcome
from py.__.misc.terminal_helper import ansi_print, get_terminal_width
from py.__.test.representation import Presenter
import sys
import thread
def choose_reporter(config):
option = config.option
if option.startserver or option.runbrowser:
from py.__.test.rsession.web import WebReporter
return WebReporter
if option.restreport:
from py.__.test.rsession.rest import RestReporter
return RestReporter
else:
if option.dist:
return RemoteReporter
else:
return LocalReporter
class TestReporter(object):
""" Simple test reporter which tracks failures
and also calls arbitrary provided function,
useful for tests
"""
def __init__(self, reportfun):
self.reportfun = reportfun
self.flag = False
def report(self, event):
if event.is_failure():
self.flag = True
self.reportfun(event)
__call__ = report
def was_failure(self):
return self.flag
class AbstractReporter(object):
def __init__(self, config, hosts):
@ -45,6 +77,11 @@ class AbstractReporter(object):
for i in excinfo.traceback:
print str(i)[2:-1]
print excinfo
# XXX reenable test before removing below line and
# run it with raise
#raise
__call__ = report
def report_unknown(self, what):
if self.config.option.verbose:
@ -152,7 +189,7 @@ class AbstractReporter(object):
self.repr_failure(event.item, event.outcome)
else:
self.out.sep('_', " ".join(event.item.listnames()))
out = outcome.Outcome(excinfo=event.excinfo)
out = outcome.SerializableOutcome(excinfo=event.excinfo)
self.repr_failure(event.item, outcome.ReprOutcome(out.make_repr()))
def gethost(self, event):
@ -260,7 +297,7 @@ class AbstractReporter(object):
# 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
lgt = get_terminal_width() - 20
# mark the function name, to be sure
to_display = len(itempath) + len(funname) + 1
if to_display > lgt:
@ -273,6 +310,9 @@ class AbstractReporter(object):
def report_Nodes(self, event):
self.nodes = event.nodes
def was_failure(self):
return len(self.failed) > 0
class RemoteReporter(AbstractReporter):
def get_item_name(self, event, colitem):
return event.host.hostname + ":" + \

10
py/test/rsession/executor.py
View File

@ -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 @@ class RunExecutor(object):
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 @@ class RunExecutor(object):
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,

2
py/test/rsession/hostmanage.py
View File

@ -5,7 +5,7 @@ import thread, threading
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

4
py/test/rsession/local.py
View File

@ -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):

35
py/test/rsession/master.py
View File

@ -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 @@ class MasterNode(object):
# 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),

126
py/test/rsession/outcome.py
View 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),)

4
py/test/rsession/rest.py
View File

@ -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):

119
py/test/rsession/rsession.py
View File

@ -8,101 +8,15 @@ import sys
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 @@ class RSession(AbstractSession):
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 @@ class RSession(AbstractSession):
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 @@ class LSession(AbstractSession):
"""
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 @@ class LSession(AbstractSession):
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)

6
py/test/rsession/slave.py
View File

@ -4,7 +4,7 @@ Node code for slaves.
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 @@ def slave_main(receive, send, path, config):
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

2
py/test/rsession/testing/test_executor.py
View File

@ -4,7 +4,7 @@ import example1
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

2
py/test/rsession/testing/test_hostmanage.py
View File

@ -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):

2
py/test/rsession/testing/test_lsession.py
View File

@ -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):

14
py/test/rsession/testing/test_master.py
View File

@ -11,8 +11,8 @@ if sys.platform == 'win32':
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 @@ def test_masternode():
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 @@ def test_sending_two_noes():
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

35
py/test/rsession/testing/test_rest.py
View File

@ -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 @@ Testing module foo/bar.py (2 items)
"""
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 @@ Testing module foo/bar.py (2 items)
'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 @@ Testing module foo/bar.py (2 items)
'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 @@ Testing module foo/bar.py (2 items)
),
]
)
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 @@ class TestRestReporter(AbstractTestReporter):
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
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\: **SKIPPED** py/test/rsession/testing/test\_slave.py/funcpass
* localhost\: **FAILED** `traceback1`_\n py/test/rsession/testing/test\_slave.py/funcpass
* localhost\: **PASSED** py/test/rsession/testing/test\_slave.py/funcpass
"""
print val
assert val == expected

20
py/test/rsession/testing/test_rsession.py
View File

@ -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 @@ def setup_module(mod):
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 @@ class TestRSessionRemote(DirSetup, BasicRsessionTest):
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")

2
py/test/rsession/testing/test_slave.py
View File

@ -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

49
py/test/rsession/web.py
View File

@ -15,7 +15,7 @@ import socket
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 @@ class ExportedMethods(BasicExternal):
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:
@ -316,6 +315,13 @@ class ExportedMethods(BasicExternal):
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__,
self.report_unknown)
@ -330,13 +336,6 @@ class ExportedMethods(BasicExternal):
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 @@ class TestHandler(BaseHTTPRequestHandler):
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 @@ class TestHandler(BaseHTTPRequestHandler):
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)

3862
py/test/rsession/webdata/source.js
View File

File diff suppressed because it is too large Load Diff

54
py/test/session.py
View File

@ -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
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 @@ class Session(object):
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. """

30
py/test/testing/test_collect.py
View File

@ -7,10 +7,6 @@ def setup_module(mod):
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 @@ def test__tryiter_ignores_failing_collectors():
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 @@ def test_check_generator_collect_problems():
"""))
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")

1
py/test/testing/test_config.py
View File

@ -2,7 +2,6 @@ from __future__ import generators
import py
from py.__.test.config import gettopdir
from py.__.test.testing.test_collect import skipboxed
def test_tmpdir():
d1 = py.test.ensuretemp('hello')

6
py/test/rsession/testing/test_outcome.py → py/test/testing/test_outcome.py
View File

@ -1,11 +1,11 @@
import py
from py.__.test.rsession.outcome import Outcome, ReprOutcome, ExcInfoRepr
from py.__.test.outcome import SerializableOutcome, ReprOutcome, ExcInfoRepr
import marshal
def test_critical_debugging_flag():
outcome = Outcome(is_critical=True)
outcome = SerializableOutcome(is_critical=True)
r = ReprOutcome(outcome.make_repr())
assert r.is_critical
@ -26,7 +26,7 @@ def test_exception_info_repr():
try:
f3()
except:
outcome = Outcome(excinfo=py.code.ExceptionInfo())
outcome = SerializableOutcome(excinfo=py.code.ExceptionInfo())
repr = outcome.make_excinfo_repr("long")
assert marshal.dumps(repr)

23
py/test/rsession/testing/test_repevent.py → py/test/testing/test_repevent.py
View File

@ -1,7 +1,7 @@
""" test reporting functionality. """
import py
from py.__.test.rsession import repevent
from py.__.test import repevent
def test_wrapcall_ok():
l = []
@ -27,10 +27,27 @@ def test_wrapcall_exception():
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
from py.__.test.reporter import RemoteReporter
for method in dir(RemoteReporter):
if method.startswith("report_") and method != "report_unknown":
assert method[len('report_'):] in repevent.__dict__
def test_repevent_failures():
from py.__.test.outcome import SerializableOutcome, ReprOutcome
assert not repevent.ReportEvent().is_failure()
assert not repevent.CallEvent(None, None, None).is_failure()
assert repevent.FailedTryiter(None, None).is_failure()
out = ReprOutcome(SerializableOutcome().make_repr())
assert not repevent.ReceivedItemOutcome(None, None, out).is_failure()
out = ReprOutcome(SerializableOutcome(skipped=True).make_repr())
assert not repevent.ReceivedItemOutcome(None, None, out).is_failure()
try:
1/0
except:
exc = py.code.ExceptionInfo()
out = ReprOutcome(SerializableOutcome(excinfo=exc).make_repr())
assert repevent.ReceivedItemOutcome(None, None, out).is_failure()

37
py/test/rsession/testing/test_reporter.py → py/test/testing/test_reporter.py
View File

@ -18,13 +18,14 @@ 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.session import AbstractSession
from py.__.test.reporter import RemoteReporter, LocalReporter, choose_reporter
from py.__.test import repevent
from py.__.test.outcome import ReprOutcome, SerializableOutcome
from py.__.test.rsession.hostmanage import HostInfo
from py.__.test.rsession.box import Box
from py.__.test.rsession.testing.basetest import BasicRsessionTest
from py.__.test.rsession.master import itemgen
import sys
from StringIO import StringIO
@ -44,10 +45,10 @@ class AbstractTestReporter(BasicRsessionTest):
except:
exc = py.code.ExceptionInfo()
outcomes = [Outcome(()),
Outcome(skipped=True),
Outcome(excinfo=exc),
Outcome()]
outcomes = [SerializableOutcome(()),
SerializableOutcome(skipped=True),
SerializableOutcome(excinfo=exc),
SerializableOutcome()]
outcomes = [ReprOutcome(outcome.make_repr()) for outcome in outcomes]
outcomes[3].signal = 11
@ -111,7 +112,7 @@ class AbstractTestReporter(BasicRsessionTest):
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)))
list(itemgen([rootcol], r.report))
cap = py.io.StdCaptureFD()
boxfun()
@ -132,7 +133,7 @@ class AbstractTestReporter(BasicRsessionTest):
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)))
list(itemgen([rootcol], r.report))
r.report(repevent.TestFinished())
return r
@ -149,6 +150,8 @@ class AbstractTestReporter(BasicRsessionTest):
cap = py.io.StdCaptureFD()
config = py.test.config._reparse([str(tmpdir)])
hosts = [HostInfo(i) for i in ["host1", "host2", "host3"]]
for host in hosts:
host.gw_remotepath = ''
r = self.reporter(config, hosts)
r.report(repevent.TestStarted(hosts, config.topdir, ["a", "b", "c"]))
for host in hosts:
@ -214,3 +217,17 @@ class TestRemoteReporter(AbstractTestReporter):
val = self._test_full_module()
assert val.find("FAILED TO LOAD MODULE: repmod/test_three.py\n"\
"\nSkipped ('reason') repmod/test_two.py") != -1
def test_reporter_choice():
from py.__.test.rsession.web import WebReporter
from py.__.test.rsession.rest import RestReporter
choices = [
(['-d'], RemoteReporter),
(['-d', '--rest'], RestReporter),
([], LocalReporter),
(['-w'], WebReporter),
(['-r'], WebReporter)]
for opts, reporter in choices:
config = py.test.config._reparse(['xxx'] + opts)
assert choose_reporter(config) is reporter