import py import sys def pytest_configure(config): if config.option.collectonly: reporter = CollectonlyReporter(config) else: reporter = TerminalReporter(config) # XXX see remote.py's XXX for attr in 'pytest_terminal_hasmarkup', 'pytest_terminal_fullwidth': if hasattr(config, attr): #print "SETTING TERMINAL OPTIONS", attr, getattr(config, attr) name = attr.split("_")[-1] assert hasattr(self.reporter._tw, name), name setattr(reporter._tw, name, getattr(config, attr)) config.pluginmanager.register(reporter) class TerminalReporter: def __init__(self, config, file=None): self.config = config self.stats = {} self.curdir = py.path.local() if file is None: file = py.std.sys.stdout self._tw = py.io.TerminalWriter(file) self.currentfspath = None self.gateway2info = {} def write_fspath_result(self, fspath, res): fspath = self.curdir.bestrelpath(fspath) if fspath != self.currentfspath: self._tw.line() relpath = self.curdir.bestrelpath(fspath) self._tw.write(relpath + " ") self.currentfspath = fspath self._tw.write(res) def write_ensure_prefix(self, prefix, extra="", **kwargs): if self.currentfspath != prefix: self._tw.line() self.currentfspath = prefix self._tw.write(prefix) if extra: self._tw.write(extra, **kwargs) self.currentfspath = -2 def ensure_newline(self): if self.currentfspath: self._tw.line() self.currentfspath = None def write_line(self, line, **markup): line = str(line) self.ensure_newline() self._tw.line(line, **markup) def write_sep(self, sep, title=None, **markup): self.ensure_newline() self._tw.sep(sep, title, **markup) def getcategoryletterword(self, rep): res = self.config.hook.pytest_report_teststatus(rep=rep) if res: return res for cat in 'skipped failed passed ???'.split(): if getattr(rep, cat, None): break return cat, self.getoutcomeletter(rep), self.getoutcomeword(rep) def getoutcomeletter(self, rep): return rep.shortrepr def getoutcomeword(self, rep): if rep.passed: return "PASS", dict(green=True) elif rep.failed: return "FAIL", dict(red=True) elif rep.skipped: return "SKIP" else: return "???", dict(red=True) def pytest_internalerror(self, excrepr): for line in str(excrepr).split("\n"): self.write_line("INTERNALERROR> " + line) def pyexecnet_gwmanage_newgateway(self, gateway, platinfo): #self.write_line("%s instantiated gateway from spec %r" %(gateway.id, gateway.spec._spec)) d = {} d['version'] = repr_pythonversion(platinfo.version_info) d['id'] = gateway.id d['spec'] = gateway.spec._spec d['platform'] = platinfo.platform if self.config.option.verbose: d['extra'] = "- " + platinfo.executable else: d['extra'] = "" d['cwd'] = platinfo.cwd infoline = ("%(id)s %(spec)s -- platform %(platform)s, " "Python %(version)s " "cwd: %(cwd)s" "%(extra)s" % d) self.write_line(infoline) self.gateway2info[gateway] = infoline def pyexecnet_gwmanage_rsyncstart(self, source, gateways): targets = ", ".join([gw.id for gw in gateways]) msg = "rsyncstart: %s -> %s" %(source, targets) if not self.config.option.verbose: msg += " # use --verbose to see rsync progress" self.write_line(msg) def pyexecnet_gwmanage_rsyncfinish(self, source, gateways): targets = ", ".join([gw.id for gw in gateways]) self.write_line("rsyncfinish: %s -> %s" %(source, targets)) def pytest_plugin_registered(self, plugin): if self.config.option.traceconfig: msg = "PLUGIN registered: %s" %(plugin,) # XXX this event may happen during setup/teardown time # which unfortunately captures our output here # which garbles our output if we use self.write_line self.write_line(msg) def pytest_testnodeready(self, node): self.write_line("%s txnode ready to receive tests" %(node.gateway.id,)) def pytest_testnodedown(self, node, error): if error: self.write_line("%s node down, error: %s" %(node.gateway.id, error)) def pytest_trace(self, category, msg): if self.config.option.debug or \ self.config.option.traceconfig and category.find("config") != -1: self.write_line("[%s] %s" %(category, msg)) def pytest_rescheduleitems(self, items): if self.config.option.debug: self.write_sep("!", "RESCHEDULING %s " %(items,)) def pytest_deselected(self, items): self.stats.setdefault('deselected', []).append(items) def pytest_itemstart(self, item, node=None): if self.config.option.dist != "no": # for dist-testing situations itemstart means we # queued the item for sending, not interesting (unless debugging) if self.config.option.debug: line = self._reportinfoline(item) extra = "" if node: extra = "-> " + str(node.gateway.id) self.write_ensure_prefix(line, extra) else: if self.config.option.verbose: line = self._reportinfoline(item) self.write_ensure_prefix(line, "") else: # ensure that the path is printed before the # 1st test of a module starts running fspath, lineno, msg = self._getreportinfo(item) self.write_fspath_result(fspath, "") def pytest_itemtestreport(self, rep): fspath = rep.item.fspath cat, letter, word = self.getcategoryletterword(rep) if isinstance(word, tuple): word, markup = word else: markup = {} self.stats.setdefault(cat, []).append(rep) if not self.config.option.verbose: fspath, lineno, msg = self._getreportinfo(rep.item) self.write_fspath_result(fspath, letter) else: line = self._reportinfoline(rep.item) if not hasattr(rep, 'node'): self.write_ensure_prefix(line, word, **markup) else: self.ensure_newline() if hasattr(rep, 'node'): self._tw.write("%s " % rep.node.gateway.id) self._tw.write(word, **markup) self._tw.write(" " + line) self.currentfspath = -2 def pytest_collectreport(self, rep): if not rep.passed: if rep.failed: self.stats.setdefault("failed", []).append(rep) msg = rep.longrepr.reprcrash.message self.write_fspath_result(rep.collector.fspath, "F") elif rep.skipped: self.stats.setdefault("skipped", []).append(rep) self.write_fspath_result(rep.collector.fspath, "S") def pytest_sessionstart(self, session): self.write_sep("=", "test session starts", bold=True) self._sessionstarttime = py.std.time.time() verinfo = ".".join(map(str, sys.version_info[:3])) msg = "python: platform %s -- Python %s" % (sys.platform, verinfo) if self.config.option.verbose or self.config.option.debug: msg += " -- " + str(sys.executable) self.write_line(msg) if self.config.option.debug or self.config.option.traceconfig: rev = py.__pkg__.getrev() self.write_line("using py lib: %s " % ( py.path.local(py.__file__).dirpath(), rev)) if self.config.option.traceconfig: plugins = [] for plugin in self.config.pluginmanager.comregistry: name = plugin.__class__.__name__ if name.endswith("Plugin"): name = name[:-6] #if name == "Conftest": # XXX get filename plugins.append(name) else: plugins.append(str(plugin)) plugins = ", ".join(plugins) self.write_line("active plugins: %s" %(plugins,)) for i, testarg in py.builtin.enumerate(self.config.args): self.write_line("test object %d: %s" %(i+1, testarg)) def pytest_sessionfinish(self, session, exitstatus, excrepr=None): self._tw.line("") if exitstatus in (0, 1, 2): self.summary_failures() self.summary_skips() self.config.hook.pytest_terminal_summary(terminalreporter=self) if excrepr is not None: self.summary_final_exc(excrepr) if exitstatus == 2: self.write_sep("!", "KEYBOARD INTERRUPT") self.summary_deselected() self.summary_stats() def pytest_looponfailinfo(self, failreports, rootdirs): if failreports: self.write_sep("#", "LOOPONFAILING", red=True) for report in failreports: try: loc = report.longrepr.reprcrash except AttributeError: loc = str(report.longrepr)[:50] self.write_line(loc, red=True) self.write_sep("#", "waiting for changes") for rootdir in rootdirs: self.write_line("### Watching: %s" %(rootdir,), bold=True) def _reportinfoline(self, item): fspath, lineno, msg = self._getreportinfo(item) if fspath: fspath = self.curdir.bestrelpath(fspath) if lineno is not None: lineno += 1 if fspath and lineno and msg: line = "%(fspath)s:%(lineno)s: %(msg)s" elif fspath and msg: line = "%(fspath)s: %(msg)s" elif fspath and lineno: line = "%(fspath)s:%(lineno)s" else: line = "[noreportinfo]" return line % locals() + " " def _getfailureheadline(self, rep): if hasattr(rep, "collector"): return str(rep.collector.fspath) else: fspath, lineno, msg = self._getreportinfo(rep.item) return msg def _getreportinfo(self, item): try: return item.__reportinfo except AttributeError: pass reportinfo = item.config.hook.pytest_report_iteminfo(item=item) # cache on item item.__reportinfo = reportinfo return reportinfo # # summaries for sessionfinish # def summary_failures(self): if 'failed' in self.stats and self.config.option.tbstyle != "no": self.write_sep("=", "FAILURES") for rep in self.stats['failed']: msg = self._getfailureheadline(rep) self.write_sep("_", msg) if hasattr(rep, 'node'): self.write_line(self.gateway2info.get( rep.node.gateway, "node %r (platinfo not found? strange)") [:self._tw.fullwidth-1]) rep.toterminal(self._tw) def summary_stats(self): session_duration = py.std.time.time() - self._sessionstarttime keys = "failed passed skipped deselected".split() parts = [] for key in keys: val = self.stats.get(key, None) if val: parts.append("%d %s" %(len(val), key)) line = ", ".join(parts) # XXX coloring self.write_sep("=", "%s in %.2f seconds" %(line, session_duration)) def summary_deselected(self): if 'deselected' in self.stats: self.write_sep("=", "%d tests deselected by %r" %( len(self.stats['deselected']), self.config.option.keyword), bold=True) def summary_skips(self): if 'skipped' in self.stats: if 'failed' not in self.stats: # or self.config.option.showskipsummary: fskips = folded_skips(self.stats['skipped']) if fskips: self.write_sep("_", "skipped test summary") for num, fspath, lineno, reason in fskips: self._tw.line("%s:%d: [%d] %s" %(fspath, lineno, num, reason)) def summary_final_exc(self, excrepr): self.write_sep("!") if self.config.option.verbose: excrepr.toterminal(self._tw) else: excrepr.reprcrash.toterminal(self._tw) class CollectonlyReporter: INDENT = " " def __init__(self, config, out=None): self.config = config if out is None: out = py.std.sys.stdout self.out = py.io.TerminalWriter(out) self.indent = "" self._failed = [] def outindent(self, line): self.out.line(self.indent + str(line)) def pytest_collectstart(self, collector): self.outindent(collector) self.indent += self.INDENT def pytest_itemstart(self, item, node=None): self.outindent(item) def pytest_collectreport(self, rep): if not rep.passed: self.outindent("!!! %s !!!" % rep.longrepr.reprcrash.message) self._failed.append(rep) self.indent = self.indent[:-len(self.INDENT)] def pytest_sessionfinish(self, session, exitstatus, excrepr=None): if self._failed: self.out.sep("!", "collection failures") for rep in self._failed: rep.toterminal(self.out) def folded_skips(skipped): d = {} for event in skipped: entry = event.longrepr.reprcrash key = entry.path, entry.lineno, entry.message d.setdefault(key, []).append(event) l = [] for key, events in d.iteritems(): l.append((len(events),) + key) return l def repr_pythonversion(v=None): if v is None: v = sys.version_info try: return "%s.%s.%s-%s-%s" % v except (TypeError, ValueError): return str(v) # =============================================================================== # # plugin tests # # =============================================================================== from py.__.test import runner class TestTerminal: def test_pass_skip_fail(self, testdir, linecomp): modcol = testdir.getmodulecol(""" import py def test_ok(): pass def test_skip(): py.test.skip("xx") def test_func(): assert 0 """) rep = TerminalReporter(modcol.config, file=linecomp.stringio) rep.config.pluginmanager.register(rep) rep.config.hook.pytest_sessionstart(session=testdir.session) for item in testdir.genitems([modcol]): ev = runner.basic_run_report(item) rep.config.hook.pytest_itemtestreport(rep=ev) linecomp.assert_contains_lines([ "*test_pass_skip_fail.py .sF" ]) rep.config.hook.pytest_sessionfinish(session=testdir.session, exitstatus=1) linecomp.assert_contains_lines([ " def test_func():", "> assert 0", "E assert 0", ]) def test_pass_skip_fail_verbose(self, testdir, linecomp): modcol = testdir.getmodulecol(""" import py def test_ok(): pass def test_skip(): py.test.skip("xx") def test_func(): assert 0 """, configargs=("-v",)) rep = TerminalReporter(modcol.config, file=linecomp.stringio) rep.config.pluginmanager.register(rep) rep.config.hook.pytest_sessionstart(session=testdir.session) items = modcol.collect() rep.config.option.debug = True # for item in items: rep.config.hook.pytest_itemstart(item=item, node=None) s = linecomp.stringio.getvalue().strip() assert s.endswith(item.name) rep.config.hook.pytest_itemtestreport(rep=runner.basic_run_report(item)) linecomp.assert_contains_lines([ "*test_pass_skip_fail_verbose.py:2: *test_ok*PASS*", "*test_pass_skip_fail_verbose.py:4: *test_skip*SKIP*", "*test_pass_skip_fail_verbose.py:6: *test_func*FAIL*", ]) rep.config.hook.pytest_sessionfinish(session=testdir.session, exitstatus=1) linecomp.assert_contains_lines([ " def test_func():", "> assert 0", "E assert 0", ]) def test_collect_fail(self, testdir, linecomp): modcol = testdir.getmodulecol("import xyz") rep = TerminalReporter(modcol.config, file=linecomp.stringio) rep.config.pluginmanager.register(rep) rep.config.hook.pytest_sessionstart(session=testdir.session) l = list(testdir.genitems([modcol])) assert len(l) == 0 linecomp.assert_contains_lines([ "*test_collect_fail.py F*" ]) rep.config.hook.pytest_sessionfinish(session=testdir.session, exitstatus=1) linecomp.assert_contains_lines([ "> import xyz", "E ImportError: No module named xyz" ]) def test_internalerror(self, testdir, linecomp): modcol = testdir.getmodulecol("def test_one(): pass") rep = TerminalReporter(modcol.config, file=linecomp.stringio) excinfo = py.test.raises(ValueError, "raise ValueError('hello')") rep.pytest_internalerror(excinfo.getrepr()) linecomp.assert_contains_lines([ "INTERNALERROR> *raise ValueError*" ]) def test_gwmanage_events(self, testdir, linecomp): modcol = testdir.getmodulecol(""" def test_one(): pass """, configargs=("-v",)) rep = TerminalReporter(modcol.config, file=linecomp.stringio) class gw1: id = "X1" spec = py.execnet.XSpec("popen") class gw2: id = "X2" spec = py.execnet.XSpec("popen") class rinfo: version_info = (2, 5, 1, 'final', 0) executable = "hello" platform = "xyz" cwd = "qwe" rep.pyexecnet_gwmanage_newgateway(gw1, rinfo) linecomp.assert_contains_lines([ "X1*popen*xyz*2.5*" ]) rep.pyexecnet_gwmanage_rsyncstart(source="hello", gateways=[gw1, gw2]) linecomp.assert_contains_lines([ "rsyncstart: hello -> X1, X2" ]) rep.pyexecnet_gwmanage_rsyncfinish(source="hello", gateways=[gw1, gw2]) linecomp.assert_contains_lines([ "rsyncfinish: hello -> X1, X2" ]) def test_writeline(self, testdir, linecomp): modcol = testdir.getmodulecol("def test_one(): pass") stringio = py.std.cStringIO.StringIO() rep = TerminalReporter(modcol.config, file=linecomp.stringio) rep.write_fspath_result(py.path.local("xy.py"), '.') rep.write_line("hello world") lines = linecomp.stringio.getvalue().split('\n') assert not lines[0] assert lines[1].endswith("xy.py .") assert lines[2] == "hello world" def test_looponfailreport(self, testdir, linecomp): modcol = testdir.getmodulecol(""" def test_fail(): assert 0 def test_fail2(): raise ValueError() """) rep = TerminalReporter(modcol.config, file=linecomp.stringio) reports = [runner.basic_run_report(x) for x in modcol.collect()] rep.pytest_looponfailinfo(reports, [modcol.config.topdir]) linecomp.assert_contains_lines([ "*test_looponfailreport.py:2: assert 0", "*test_looponfailreport.py:4: ValueError*", "*waiting*", "*%s*" % (modcol.config.topdir), ]) def test_tb_option(self, testdir, linecomp): # XXX usage of testdir for tbopt in ["long", "short", "no"]: print 'testing --tb=%s...' % tbopt modcol = testdir.getmodulecol(""" import py def g(): raise IndexError def test_func(): print 6*7 g() # --calling-- """, configargs=("--tb=%s" % tbopt,)) rep = TerminalReporter(modcol.config, file=linecomp.stringio) rep.config.pluginmanager.register(rep) rep.config.hook.pytest_sessionstart(session=testdir.session) for item in testdir.genitems([modcol]): rep.config.hook.pytest_itemtestreport( rep=runner.basic_run_report(item)) rep.config.hook.pytest_sessionfinish(session=testdir.session, exitstatus=1) s = linecomp.stringio.getvalue() if tbopt == "long": print s assert 'print 6*7' in s else: assert 'print 6*7' not in s if tbopt != "no": assert '--calling--' in s assert 'IndexError' in s else: assert 'FAILURES' not in s assert '--calling--' not in s assert 'IndexError' not in s linecomp.stringio.truncate(0) def test_show_path_before_running_test(self, testdir, linecomp): item = testdir.getitem("def test_func(): pass") tr = TerminalReporter(item.config, file=linecomp.stringio) item.config.pluginmanager.register(tr) tr.config.hook.pytest_itemstart(item=item) linecomp.assert_contains_lines([ "*test_show_path_before_running_test.py*" ]) def test_itemreport_reportinfo(self, testdir, linecomp): testdir.makeconftest(""" import py class Function(py.test.collect.Function): def reportinfo(self): return "ABCDE", 42, "custom" """) item = testdir.getitem("def test_func(): pass") tr = TerminalReporter(item.config, file=linecomp.stringio) item.config.pluginmanager.register(tr) tr.config.hook.pytest_itemstart(item=item) linecomp.assert_contains_lines([ "*ABCDE " ]) tr.config.option.verbose = True tr.config.hook.pytest_itemstart(item=item) linecomp.assert_contains_lines([ "*ABCDE:43: custom*" ]) def test_itemreport_pytest_report_iteminfo(self, testdir, linecomp): item = testdir.getitem("def test_func(): pass") class Plugin: def pytest_report_iteminfo(self, item): return "FGHJ", 42, "custom" item.config.pluginmanager.register(Plugin()) tr = TerminalReporter(item.config, file=linecomp.stringio) item.config.pluginmanager.register(tr) tr.config.hook.pytest_itemstart(item=item) linecomp.assert_contains_lines([ "*FGHJ " ]) tr.config.option.verbose = True tr.config.hook.pytest_itemstart(item=item) linecomp.assert_contains_lines([ "*FGHJ:43: custom*" ]) def pseudo_keyboard_interrupt(self, testdir, linecomp, verbose=False): modcol = testdir.getmodulecol(""" def test_foobar(): assert 0 def test_spamegg(): import py; py.test.skip('skip me please!') def test_interrupt_me(): raise KeyboardInterrupt # simulating the user """, configargs=("-v",)*verbose) #""", configargs=("--showskipsummary",) + ("-v",)*verbose) rep = TerminalReporter(modcol.config, file=linecomp.stringio) modcol.config.pluginmanager.register(rep) modcol.config.hook.pytest_sessionstart(session=testdir.session) try: for item in testdir.genitems([modcol]): modcol.config.hook.pytest_itemtestreport( rep=runner.basic_run_report(item)) except KeyboardInterrupt: excinfo = py.code.ExceptionInfo() else: py.test.fail("no KeyboardInterrupt??") s = linecomp.stringio.getvalue() if not verbose: assert s.find("_keyboard_interrupt.py Fs") != -1 modcol.config.hook.pytest_sessionfinish( session=testdir.session, exitstatus=2, excrepr=excinfo.getrepr()) text = linecomp.stringio.getvalue() linecomp.assert_contains_lines([ " def test_foobar():", "> assert 0", "E assert 0", ]) #assert "Skipped: 'skip me please!'" in text assert "_keyboard_interrupt.py:6: KeyboardInterrupt" in text see_details = "raise KeyboardInterrupt # simulating the user" in text assert see_details == verbose def test_keyboard_interrupt(self, testdir, linecomp): self.pseudo_keyboard_interrupt(testdir, linecomp) def test_verbose_keyboard_interrupt(self, testdir, linecomp): self.pseudo_keyboard_interrupt(testdir, linecomp, verbose=True) def test_skip_reasons_folding(self): class longrepr: class reprcrash: path = 'xyz' lineno = 3 message = "justso" ev1 = runner.CollectReport(None, None) ev1.when = "execute" ev1.skipped = True ev1.longrepr = longrepr ev2 = runner.ItemTestReport(None, excinfo=longrepr) ev2.skipped = True l = folded_skips([ev1, ev2]) assert len(l) == 1 num, fspath, lineno, reason = l[0] assert num == 2 assert fspath == longrepr.reprcrash.path assert lineno == longrepr.reprcrash.lineno assert reason == longrepr.reprcrash.message class TestCollectonly: def test_collectonly_basic(self, testdir, linecomp): modcol = testdir.getmodulecol(configargs=['--collectonly'], source=""" def test_func(): pass """) rep = CollectonlyReporter(modcol.config, out=linecomp.stringio) modcol.config.pluginmanager.register(rep) indent = rep.indent rep.config.hook.pytest_collectstart(collector=modcol) linecomp.assert_contains_lines([ "" ]) item = modcol.join("test_func") rep.config.hook.pytest_itemstart(item=item) linecomp.assert_contains_lines([ " ", ]) rep.config.hook.pytest_collectreport( rep=runner.CollectReport(modcol, [], excinfo=None)) assert rep.indent == indent def test_collectonly_skipped_module(self, testdir, linecomp): modcol = testdir.getmodulecol(configargs=['--collectonly'], source=""" import py py.test.skip("nomod") """) rep = CollectonlyReporter(modcol.config, out=linecomp.stringio) modcol.config.pluginmanager.register(rep) cols = list(testdir.genitems([modcol])) assert len(cols) == 0 linecomp.assert_contains_lines(""" !!! Skipped: 'nomod' !!! """) def test_collectonly_failed_module(self, testdir, linecomp): modcol = testdir.getmodulecol(configargs=['--collectonly'], source=""" raise ValueError(0) """) rep = CollectonlyReporter(modcol.config, out=linecomp.stringio) modcol.config.pluginmanager.register(rep) cols = list(testdir.genitems([modcol])) assert len(cols) == 0 linecomp.assert_contains_lines(""" !!! ValueError: 0 !!! """) def test_repr_python_version(monkeypatch): monkeypatch.setattr(sys, 'version_info', (2, 5, 1, 'final', 0)) assert repr_pythonversion() == "2.5.1-final-0" py.std.sys.version_info = x = (2,3) assert repr_pythonversion() == str(x) def test_generic(plugintester): plugintester.hookcheck() plugintester.hookcheck(cls=TerminalReporter) plugintester.hookcheck(cls=CollectonlyReporter)