diff --git a/py/test/rsession/rest.py.merge.tmp b/py/test/rsession/rest.py.merge.tmp new file mode 100644 index 000000000..caaa4f3a8 --- /dev/null +++ b/py/test/rsession/rest.py.merge.tmp @@ -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) +