768 lines
		
	
	
		
			28 KiB
		
	
	
	
		
			Python
		
	
	
	
			
		
		
	
	
			768 lines
		
	
	
		
			28 KiB
		
	
	
	
		
			Python
		
	
	
	
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_runtest_logreport(self, rep):
 | 
						|
        if rep.passed and rep.when in ("setup", "teardown"):
 | 
						|
            return 
 | 
						|
        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 <rev %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_internalerror(self, excrepr):
 | 
						|
        for line in str(excrepr).split("\n"):
 | 
						|
            self.out.line("INTERNALERROR> " + 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 
 | 
						|
#
 | 
						|
# ===============================================================================
 | 
						|
 | 
						|
import pytest_runner as runner # XXX 
 | 
						|
 | 
						|
def basic_run_report(item):
 | 
						|
    return runner.call_and_report(item, "call", log=False)
 | 
						|
 | 
						|
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 = basic_run_report(item) 
 | 
						|
            rep.config.hook.pytest_runtest_logreport(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_runtest_logreport(rep=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 = [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_runtest_logreport(
 | 
						|
                    rep=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_runtest_logreport(
 | 
						|
                    rep=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([
 | 
						|
           "<Module 'test_collectonly_basic.py'>"
 | 
						|
        ])
 | 
						|
        item = modcol.join("test_func")
 | 
						|
        rep.config.hook.pytest_itemstart(item=item)
 | 
						|
        linecomp.assert_contains_lines([
 | 
						|
           "  <Function 'test_func'>", 
 | 
						|
        ])
 | 
						|
        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("""
 | 
						|
            <Module 'test_collectonly_skipped_module.py'>
 | 
						|
              !!! 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("""
 | 
						|
            <Module 'test_collectonly_failed_module.py'>
 | 
						|
              !!! ValueError: 0 !!!
 | 
						|
        """)
 | 
						|
 | 
						|
    def test_collectonly_fatal(self, testdir):
 | 
						|
        p1 = testdir.makeconftest("""
 | 
						|
            def pytest_collectstart(collector):
 | 
						|
                assert 0, "urgs" 
 | 
						|
        """)
 | 
						|
        result = testdir.runpytest("--collectonly") 
 | 
						|
        result.stdout.fnmatch_lines([
 | 
						|
            "*INTERNAL*args*"
 | 
						|
        ])
 | 
						|
        assert result.ret == 3
 | 
						|
 | 
						|
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) 
 | 
						|
 |