[svn r37272] merging of file:///svn/py/branch/config/py/test/rsession/rest.py
revisions 36936 to 37267:
------------------------------------------------------------------------
r37178 | fijal | 2007-01-23 11:38:39 +0100 (Tue, 23 Jan 2007) | 2 lines
Make it a bit more working.
------------------------------------------------------------------------
r36937 | fijal | 2007-01-18 14:22:54 +0100 (Thu, 18 Jan 2007) | 2 lines
Create a branch for further config cleanups.
------------------------------------------------------------------------
--HG--
branch : trunk
This commit is contained in:
hpk
parent
e94a602e59
commit
6dbd42b60e
|
|
@ -0,0 +1,276 @@
|
|||
|
||||
""" Rest reporting stuff
|
||||
"""
|
||||
|
||||
import py
|
||||
import sys
|
||||
from StringIO import StringIO
|
||||
from py.__.test.rsession.reporter import AbstractReporter
|
||||
from py.__.test.rsession import report
|
||||
from py.__.rest.rst import *
|
||||
|
||||
class RestReporter(AbstractReporter):
|
||||
linkwriter = None
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(RestReporter, self).__init__(*args, **kwargs)
|
||||
self.rest = Rest()
|
||||
self.traceback_num = 0
|
||||
|
||||
def get_linkwriter(self):
|
||||
if self.linkwriter is None:
|
||||
try:
|
||||
self.linkwriter = self.config.getvalue('linkwriter')
|
||||
except KeyError:
|
||||
print >>sys.stderr, ('no linkwriter configured, using default '
|
||||
'one')
|
||||
self.linkwriter = RelLinkWriter()
|
||||
return self.linkwriter
|
||||
|
||||
def report_unknown(self, what):
|
||||
if self.config.option.verbose:
|
||||
self.add_rest(Paragraph("Unknown report: %s" % what))
|
||||
|
||||
def gethost(self, item):
|
||||
if item.channel:
|
||||
return item.channel.gateway.host
|
||||
return self.hosts[0]
|
||||
|
||||
def report_SendItem(self, item):
|
||||
address = self.gethost(item)
|
||||
if self.config.option.verbose:
|
||||
self.add_rest(Paragraph('sending item %s to %s' % (item.item,
|
||||
address)))
|
||||
|
||||
def report_HostRSyncing(self, item):
|
||||
self.add_rest(LiteralBlock('%10s: RSYNC ==> %s' % (item.host.hostname[:10],
|
||||
item.remoterootpath)))
|
||||
|
||||
def report_HostReady(self, item):
|
||||
self.add_rest(LiteralBlock('%10s: READY' % (item.host.hostname[:10],)))
|
||||
|
||||
def report_TestStarted(self, event):
|
||||
txt = "Running tests on hosts: %s" % ", ".join(event.hosts)
|
||||
self.add_rest(Title(txt, abovechar='=', belowchar='='))
|
||||
self.timestart = event.timestart
|
||||
|
||||
def report_TestFinished(self, item):
|
||||
self.timeend = item.timeend
|
||||
self.summary()
|
||||
return len(self.failed_tests_outcome) > 0
|
||||
|
||||
def report_ImmediateFailure(self, item):
|
||||
pass
|
||||
|
||||
def report_ItemStart(self, event):
|
||||
item = event.item
|
||||
if isinstance(item, py.test.collect.Module):
|
||||
lgt = len(list(item.tryiter()))
|
||||
lns = item.listnames()[1:]
|
||||
name = "/".join(lns)
|
||||
link = self.get_linkwriter().get_link(self.get_rootpath(item),
|
||||
item.fspath)
|
||||
if link:
|
||||
name = Link(name, link)
|
||||
txt = 'Testing module %s (%d items)' % (name, lgt)
|
||||
self.add_rest(Title('Testing module', name, '(%d items)' % (lgt,),
|
||||
belowchar='-'))
|
||||
|
||||
def get_rootpath(self, item):
|
||||
root = item.parent
|
||||
while root.parent is not None:
|
||||
root = root.parent
|
||||
return root.fspath
|
||||
|
||||
def print_summary(self, total, skipped_str, failed_str):
|
||||
self.skips()
|
||||
self.failures()
|
||||
|
||||
txt = "%d tests run%s%s in %.2fs (rsync: %.2f)" % \
|
||||
(total, skipped_str, failed_str, self.timeend - self.timestart,
|
||||
self.timersync - self.timestart)
|
||||
self.add_rest(Title(txt, belowchar='-'))
|
||||
|
||||
# since we're rendering each item, the links haven't been rendered
|
||||
# yet
|
||||
self.out.write(self.rest.render_links())
|
||||
|
||||
def report_ReceivedItemOutcome(self, event):
|
||||
host = self.gethost(event)
|
||||
if event.outcome.passed:
|
||||
status = [Strong("PASSED")]
|
||||
self.passed[host] += 1
|
||||
elif event.outcome.skipped:
|
||||
status = [Strong("SKIPPED")]
|
||||
self.skipped_tests_outcome.append(event)
|
||||
self.skipped[host] += 1
|
||||
else:
|
||||
status = [Strong("FAILED"),
|
||||
InternalLink("traceback%d" % self.traceback_num)]
|
||||
self.traceback_num += 1
|
||||
self.failed[host] += 1
|
||||
self.failed_tests_outcome.append(event)
|
||||
# we'll take care of them later
|
||||
itempath = self.get_path_from_item(event.item)
|
||||
status.append(Text(itempath))
|
||||
hostname = host.hostname
|
||||
self.add_rest(ListItem(Text("%10s:" % (hostname[:10],)), *status))
|
||||
|
||||
def skips(self):
|
||||
# XXX hrmph, copied code
|
||||
texts = {}
|
||||
for event in self.skipped_tests_outcome:
|
||||
colitem = event.item
|
||||
if isinstance(event, report.ReceivedItemOutcome):
|
||||
outcome = event.outcome
|
||||
text = outcome.skipped
|
||||
itemname = self.get_item_name(event, colitem)
|
||||
elif isinstance(event, report.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.add_rest(Title('Reasons for skipped tests:', belowchar='+'))
|
||||
for text, items in texts.items():
|
||||
for item in items:
|
||||
self.add_rest(ListItem('%s: %s' % (item, text)))
|
||||
|
||||
def get_host(self, event):
|
||||
return event.channel.gateway.host
|
||||
|
||||
def failures(self):
|
||||
self.traceback_num = 0
|
||||
tbstyle = self.config.option.tbstyle
|
||||
if self.failed_tests_outcome:
|
||||
self.add_rest(Title('Exceptions:', belowchar='+'))
|
||||
for i, event in enumerate(self.failed_tests_outcome):
|
||||
if i > 0:
|
||||
self.add_rest(Transition())
|
||||
if isinstance(event, report.ReceivedItemOutcome):
|
||||
host = self.get_host(event)
|
||||
itempath = self.get_path_from_item(event.item)
|
||||
root = self.get_rootpath(event.item)
|
||||
link = self.get_linkwriter().get_link(root, event.item.fspath)
|
||||
t = Title(belowchar='+')
|
||||
if link:
|
||||
t.add(Link(itempath, link))
|
||||
else:
|
||||
t.add(Text(itempath))
|
||||
t.add(Text('on %s' % (host.hostname,)))
|
||||
self.add_rest(t)
|
||||
if event.outcome.signal:
|
||||
self.repr_signal(event.item, event.outcome)
|
||||
else:
|
||||
self.repr_failure(event.item, event.outcome, tbstyle)
|
||||
else:
|
||||
itempath = self.get_path_from_item(event.item)
|
||||
root = self.get_rootpath(event.item)
|
||||
link = self.get_linkwriter().get_link(root, event.item.fspath)
|
||||
t = Title(abovechar='+', belowchar='+')
|
||||
if link:
|
||||
t.add(Link(itempath, link))
|
||||
else:
|
||||
t.add(Text(itempath))
|
||||
out = outcome.Outcome(excinfo=event.excinfo)
|
||||
self.repr_failure(event.item,
|
||||
outcome.ReprOutcome(out.make_repr()),
|
||||
tbstyle)
|
||||
|
||||
def repr_signal(self, item, outcome):
|
||||
signal = outcome.signal
|
||||
self.add_rest(Title('Received signal: %d' % (outcome.signal,),
|
||||
abovechar='+', belowchar='+'))
|
||||
if outcome.stdout.strip():
|
||||
self.add_rest(Paragraph('Captured process stdout:'))
|
||||
self.add_rest(LiteralBlock(outcome.stdout))
|
||||
if outcome.stderr.strip():
|
||||
self.add_rest(Paragraph('Captured process stderr:'))
|
||||
self.add_rest(LiteralBlock(outcome.stderr))
|
||||
|
||||
def repr_failure(self, item, outcome, style):
|
||||
excinfo = outcome.excinfo
|
||||
traceback = excinfo.traceback
|
||||
if not traceback:
|
||||
self.add_rest(Paragraph('empty traceback from item %r' % (item,)))
|
||||
return
|
||||
self.repr_traceback(item, excinfo, traceback, style)
|
||||
if outcome.stdout:
|
||||
self.add_rest(Title('Captured process stdout:', abovechar='+',
|
||||
belowchar='+'))
|
||||
self.add_rest(LiteralBlock(outcome.stdout))
|
||||
if outcome.stderr:
|
||||
self.add_rest(Title('Captured process stderr:', abovechar='+',
|
||||
belowchar='+'))
|
||||
self.add_rest(LiteralBlock(outcome.stderr))
|
||||
|
||||
def repr_traceback(self, item, excinfo, traceback, style):
|
||||
root = self.get_rootpath(item)
|
||||
self.add_rest(LinkTarget('traceback%d' % self.traceback_num, ""))
|
||||
self.traceback_num += 1
|
||||
if style == 'long':
|
||||
for entry in traceback:
|
||||
link = self.get_linkwriter().get_link(root,
|
||||
py.path.local(entry.path))
|
||||
if link:
|
||||
self.add_rest(Title(Link(entry.path, link),
|
||||
'line %d' % (entry.lineno,),
|
||||
belowchar='+', abovechar='+'))
|
||||
else:
|
||||
self.add_rest(Title('%s line %d' % (entry.path,
|
||||
entry.lineno,),
|
||||
belowchar='+', abovechar='+'))
|
||||
self.add_rest(LiteralBlock(self.prepare_source(entry.relline,
|
||||
entry.source)))
|
||||
elif style == 'short':
|
||||
text = []
|
||||
for entry in traceback:
|
||||
text.append('%s line %d' % (entry.path, entry.lineno))
|
||||
text.append(' %s' % (entry.source.strip(),))
|
||||
self.add_rest(LiteralBlock('\n'.join(text)))
|
||||
self.add_rest(Title(excinfo.typename, belowchar='+'))
|
||||
self.add_rest(LiteralBlock(excinfo.value))
|
||||
|
||||
def prepare_source(self, relline, source):
|
||||
text = []
|
||||
for num, line in enumerate(source.split('\n')):
|
||||
if num == relline:
|
||||
text.append('>>> %s' % (line,))
|
||||
else:
|
||||
text.append(' %s' % (line,))
|
||||
return '\n'.join(text)
|
||||
|
||||
def add_rest(self, item):
|
||||
self.rest.add(item)
|
||||
self.out.write('%s\n\n' % (item.text(),))
|
||||
|
||||
def get_path_from_item(self, item):
|
||||
lns = item.listnames()[1:]
|
||||
for i, ln in enumerate(lns):
|
||||
if i > 0 and ln != '()':
|
||||
lns[i] = '/%s' % (ln,)
|
||||
itempath = ''.join(lns)
|
||||
return itempath
|
||||
|
||||
class AbstractLinkWriter(object):
|
||||
def get_link(self, base, path):
|
||||
pass
|
||||
|
||||
class NoLinkWriter(AbstractLinkWriter):
|
||||
def get_link(self, base, path):
|
||||
return ''
|
||||
|
||||
class LinkWriter(AbstractLinkWriter):
|
||||
def __init__(self, baseurl):
|
||||
self.baseurl = baseurl
|
||||
|
||||
def get_link(self, base, path):
|
||||
relpath = path.relto(base)
|
||||
return self.baseurl + relpath
|
||||
|
||||
class RelLinkWriter(AbstractLinkWriter):
|
||||
def get_link(self, base, path):
|
||||
return path.relto(base)
|
||||
|
||||
Loading…
Reference in New Issue