test_ok1/_pytest/pytester.py

""" (disabled by default) support for testing pytest and pytest plugins. """
import inspect
import sys
import os
import codecs
import re
import time
import platform
from fnmatch import fnmatch
import subprocess

import py
import pytest
from py.builtin import print_
from _pytest.core import HookRelay

from _pytest.main import Session, EXIT_OK


def get_public_names(l):
    """Only return names from iterator l without a leading underscore."""
    return [x for x in l if x[0] != "_"]


def pytest_addoption(parser):
    group = parser.getgroup("pylib")
    group.addoption('--no-tools-on-path',
           action="store_true", dest="notoolsonpath", default=False,
           help=("discover tools on PATH instead of going through py.cmdline.")
    )

def pytest_configure(config):
    # This might be called multiple times. Only take the first.
    global _pytest_fullpath
    try:
        _pytest_fullpath
    except NameError:
        _pytest_fullpath = os.path.abspath(pytest.__file__.rstrip("oc"))
        _pytest_fullpath = _pytest_fullpath.replace("$py.class", ".py")

def pytest_funcarg___pytest(request):
    return PytestArg(request)

class PytestArg:
    def __init__(self, request):
        self.request = request

    def gethookrecorder(self, hook):
        hookrecorder = HookRecorder(hook._pm)
        hookrecorder.start_recording(hook._hookspecs)
        self.request.addfinalizer(hookrecorder.finish_recording)
        return hookrecorder

class ParsedCall:
    def __init__(self, name, locals):
        assert '_name' not in locals
        self.__dict__.update(locals)
        self.__dict__.pop('self')
        self._name = name

    def __repr__(self):
        d = self.__dict__.copy()
        del d['_name']
        return "<ParsedCall %r(**%r)>" %(self._name, d)

class HookRecorder:
    def __init__(self, pluginmanager):
        self._pluginmanager = pluginmanager
        self.calls = []
        self._recorders = {}

    def start_recording(self, hookspecs):
        if not isinstance(hookspecs, (list, tuple)):
            hookspecs = [hookspecs]
        for hookspec in hookspecs:
            assert hookspec not in self._recorders
            class RecordCalls:
                _recorder = self
            for name, method in vars(hookspec).items():
                if name[0] != "_":
                    setattr(RecordCalls, name, self._makecallparser(method))
            recorder = RecordCalls()
            self._recorders[hookspec] = recorder
            self._pluginmanager.register(recorder)
        self.hook = HookRelay(hookspecs, pm=self._pluginmanager,
            prefix="pytest_")

    def finish_recording(self):
        for recorder in self._recorders.values():
            if self._pluginmanager.isregistered(recorder):
                self._pluginmanager.unregister(recorder)
        self._recorders.clear()

    def _makecallparser(self, method):
        name = method.__name__
        args, varargs, varkw, default = inspect.getargspec(method)
        if not args or args[0] != "self":
            args.insert(0, 'self')
        fspec = inspect.formatargspec(args, varargs, varkw, default)
        # we use exec because we want to have early type
        # errors on wrong input arguments, using
        # *args/**kwargs delays this and gives errors
        # elsewhere
        exec (py.code.compile("""
            def %(name)s%(fspec)s:
                        self._recorder.calls.append(
                            ParsedCall(%(name)r, locals()))
        """ % locals()))
        return locals()[name]

    def getcalls(self, names):
        if isinstance(names, str):
            names = names.split()
        for name in names:
            for cls in self._recorders:
                if name in vars(cls):
                    break
            else:
                raise ValueError("callname %r not found in %r" %(
                name, self._recorders.keys()))
        l = []
        for call in self.calls:
            if call._name in names:
                l.append(call)
        return l

    def contains(self, entries):
        __tracebackhide__ = True
        i = 0
        entries = list(entries)
        backlocals = sys._getframe(1).f_locals
        while entries:
            name, check = entries.pop(0)
            for ind, call in enumerate(self.calls[i:]):
                if call._name == name:
                    print_("NAMEMATCH", name, call)
                    if eval(check, backlocals, call.__dict__):
                        print_("CHECKERMATCH", repr(check), "->", call)
                    else:
                        print_("NOCHECKERMATCH", repr(check), "-", call)
                        continue
                    i += ind + 1
                    break
                print_("NONAMEMATCH", name, "with", call)
            else:
                pytest.fail("could not find %r check %r" % (name, check))

    def popcall(self, name):
        __tracebackhide__ = True
        for i, call in enumerate(self.calls):
            if call._name == name:
                del self.calls[i]
                return call
        lines = ["could not find call %r, in:" % (name,)]
        lines.extend(["  %s" % str(x) for x in self.calls])
        pytest.fail("\n".join(lines))

    def getcall(self, name):
        l = self.getcalls(name)
        assert len(l) == 1, (name, l)
        return l[0]


def pytest_funcarg__linecomp(request):
    return LineComp()

def pytest_funcarg__LineMatcher(request):
    return LineMatcher

def pytest_funcarg__testdir(request):
    tmptestdir = TmpTestdir(request)
    return tmptestdir

rex_outcome = re.compile("(\d+) (\w+)")
class RunResult:
    def __init__(self, ret, outlines, errlines, duration):
        self.ret = ret
        self.outlines = outlines
        self.errlines = errlines
        self.stdout = LineMatcher(outlines)
        self.stderr = LineMatcher(errlines)
        self.duration = duration

    def parseoutcomes(self):
        for line in reversed(self.outlines):
            if 'seconds' in line:
                outcomes = rex_outcome.findall(line)
                if outcomes:
                    d = {}
                    for num, cat in outcomes:
                        d[cat] = int(num)
                    return d

class TmpTestdir:
    def __init__(self, request):
        self.request = request
        self.Config = request.config.__class__
        self._pytest = request.getfuncargvalue("_pytest")
        # XXX remove duplication with tmpdir plugin
        basetmp = request.config._tmpdirhandler.ensuretemp("testdir")
        name = request.function.__name__
        for i in range(100):
            try:
                tmpdir = basetmp.mkdir(name + str(i))
            except py.error.EEXIST:
                continue
            break
        self.tmpdir = tmpdir
        self.plugins = []
        self._syspathremove = []
        self.chdir() # always chdir
        self.request.addfinalizer(self.finalize)

    def __repr__(self):
        return "<TmpTestdir %r>" % (self.tmpdir,)

    def finalize(self):
        for p in self._syspathremove:
            sys.path.remove(p)
        if hasattr(self, '_olddir'):
            self._olddir.chdir()
        # delete modules that have been loaded from tmpdir
        for name, mod in list(sys.modules.items()):
            if mod:
                fn = getattr(mod, '__file__', None)
                if fn and fn.startswith(str(self.tmpdir)):
                    del sys.modules[name]

    def getreportrecorder(self, obj):
        if hasattr(obj, 'config'):
            obj = obj.config
        if hasattr(obj, 'hook'):
            obj = obj.hook
        assert hasattr(obj, '_hookspecs'), obj
        reprec = ReportRecorder(obj)
        reprec.hookrecorder = self._pytest.gethookrecorder(obj)
        reprec.hook = reprec.hookrecorder.hook
        return reprec

    def chdir(self):
        old = self.tmpdir.chdir()
        if not hasattr(self, '_olddir'):
            self._olddir = old

    def _makefile(self, ext, args, kwargs):
        items = list(kwargs.items())
        if args:
            source = py.builtin._totext("\n").join(
                map(py.builtin._totext, args)) + py.builtin._totext("\n")
            basename = self.request.function.__name__
            items.insert(0, (basename, source))
        ret = None
        for name, value in items:
            p = self.tmpdir.join(name).new(ext=ext)
            source = py.code.Source(value)
            def my_totext(s, encoding="utf-8"):
                if py.builtin._isbytes(s):
                    s = py.builtin._totext(s, encoding=encoding)
                return s
            source_unicode = "\n".join([my_totext(line) for line in source.lines])
            source = py.builtin._totext(source_unicode)
            content = source.strip().encode("utf-8") # + "\n"
            #content = content.rstrip() + "\n"
            p.write(content, "wb")
            if ret is None:
                ret = p
        return ret


    def makefile(self, ext, *args, **kwargs):
        return self._makefile(ext, args, kwargs)

    def makeconftest(self, source):
        return self.makepyfile(conftest=source)

    def makeini(self, source):
        return self.makefile('.ini', tox=source)

    def getinicfg(self, source):
        p = self.makeini(source)
        return py.iniconfig.IniConfig(p)['pytest']

    def makepyfile(self, *args, **kwargs):
        return self._makefile('.py', args, kwargs)

    def maketxtfile(self, *args, **kwargs):
        return self._makefile('.txt', args, kwargs)

    def syspathinsert(self, path=None):
        if path is None:
            path = self.tmpdir
        sys.path.insert(0, str(path))
        self._syspathremove.append(str(path))

    def mkdir(self, name):
        return self.tmpdir.mkdir(name)

    def mkpydir(self, name):
        p = self.mkdir(name)
        p.ensure("__init__.py")
        return p

    Session = Session
    def getnode(self, config, arg):
        session = Session(config)
        assert '::' not in str(arg)
        p = py.path.local(arg)
        x = session.fspath.bestrelpath(p)
        config.hook.pytest_sessionstart(session=session)
        res = session.perform_collect([x], genitems=False)[0]
        config.hook.pytest_sessionfinish(session=session, exitstatus=EXIT_OK)
        return res

    def getpathnode(self, path):
        config = self.parseconfigure(path)
        session = Session(config)
        x = session.fspath.bestrelpath(path)
        config.hook.pytest_sessionstart(session=session)
        res = session.perform_collect([x], genitems=False)[0]
        config.hook.pytest_sessionfinish(session=session, exitstatus=EXIT_OK)
        return res

    def genitems(self, colitems):
        session = colitems[0].session
        result = []
        for colitem in colitems:
            result.extend(session.genitems(colitem))
        return result

    def runitem(self, source):
        # used from runner functional tests
        item = self.getitem(source)
        # the test class where we are called from wants to provide the runner
        testclassinstance = self.request.instance
        runner = testclassinstance.getrunner()
        return runner(item)

    def inline_runsource(self, source, *cmdlineargs):
        p = self.makepyfile(source)
        l = list(cmdlineargs) + [p]
        return self.inline_run(*l)

    def inline_runsource1(self, *args):
        args = list(args)
        source = args.pop()
        p = self.makepyfile(source)
        l = list(args) + [p]
        reprec = self.inline_run(*l)
        reports = reprec.getreports("pytest_runtest_logreport")
        assert len(reports) == 3, reports # setup/call/teardown
        return reports[1]

    def inline_genitems(self, *args):
        return self.inprocess_run(list(args) + ['--collectonly'])

    def inline_run(self, *args):
        items, rec = self.inprocess_run(args)
        return rec

    def inprocess_run(self, args, plugins=None):
        rec = []
        items = []
        class Collect:
            def pytest_configure(x, config):
                rec.append(self.getreportrecorder(config))
            def pytest_itemcollected(self, item):
                items.append(item)
        if not plugins:
            plugins = []
        plugins.append(Collect())
        ret = pytest.main(list(args), plugins=plugins)
        reprec = rec[0]
        reprec.ret = ret
        assert len(rec) == 1
        return items, reprec

    def parseconfig(self, *args):
        args = [str(x) for x in args]
        for x in args:
            if str(x).startswith('--basetemp'):
                break
        else:
            args.append("--basetemp=%s" % self.tmpdir.dirpath('basetemp'))
        import _pytest.config
        config = _pytest.config._prepareconfig(args, self.plugins)
        # we don't know what the test will do with this half-setup config
        # object and thus we make sure it gets unconfigured properly in any
        # case (otherwise capturing could still be active, for example)
        def ensure_unconfigure():
            if hasattr(config.pluginmanager, "_config"):
                config.pluginmanager.do_unconfigure(config)
            config.pluginmanager.ensure_shutdown()

        self.request.addfinalizer(ensure_unconfigure)
        return config

    def parseconfigure(self, *args):
        config = self.parseconfig(*args)
        config.do_configure()
        self.request.addfinalizer(lambda:
        config.do_unconfigure())
        return config

    def getitem(self,  source, funcname="test_func"):
        items = self.getitems(source)
        for item in items:
            if item.name == funcname:
                return item
        assert 0, "%r item not found in module:\n%s\nitems: %s" %(
                  funcname, source, items)

    def getitems(self,  source):
        modcol = self.getmodulecol(source)
        return self.genitems([modcol])

    def getmodulecol(self,  source, configargs=(), withinit=False):
        kw = {self.request.function.__name__: py.code.Source(source).strip()}
        path = self.makepyfile(**kw)
        if withinit:
            self.makepyfile(__init__ = "#")
        self.config = config = self.parseconfigure(path, *configargs)
        node = self.getnode(config, path)
        return node

    def collect_by_name(self, modcol, name):
        for colitem in modcol._memocollect():
            if colitem.name == name:
                return colitem

    def popen(self, cmdargs, stdout, stderr, **kw):
        env = os.environ.copy()
        env['PYTHONPATH'] = os.pathsep.join(filter(None, [
            str(os.getcwd()), env.get('PYTHONPATH', '')]))
        kw['env'] = env
        return subprocess.Popen(cmdargs,
                                stdout=stdout, stderr=stderr, **kw)

    def run(self, *cmdargs):
        return self._run(*cmdargs)

    def _run(self, *cmdargs):
        cmdargs = [str(x) for x in cmdargs]
        p1 = self.tmpdir.join("stdout")
        p2 = self.tmpdir.join("stderr")
        print_("running", cmdargs, "curdir=", py.path.local())
        f1 = codecs.open(str(p1), "w", encoding="utf8")
        f2 = codecs.open(str(p2), "w", encoding="utf8")
        try:
            now = time.time()
            popen = self.popen(cmdargs, stdout=f1, stderr=f2,
                close_fds=(sys.platform != "win32"))
            ret = popen.wait()
        finally:
            f1.close()
            f2.close()
        f1 = codecs.open(str(p1), "r", encoding="utf8")
        f2 = codecs.open(str(p2), "r", encoding="utf8")
        try:
            out = f1.read().splitlines()
            err = f2.read().splitlines()
        finally:
            f1.close()
            f2.close()
        self._dump_lines(out, sys.stdout)
        self._dump_lines(err, sys.stderr)
        return RunResult(ret, out, err, time.time()-now)

    def _dump_lines(self, lines, fp):
        try:
            for line in lines:
                py.builtin.print_(line, file=fp)
        except UnicodeEncodeError:
            print("couldn't print to %s because of encoding" % (fp,))

    def runpybin(self, scriptname, *args):
        fullargs = self._getpybinargs(scriptname) + args
        return self.run(*fullargs)

    def _getpybinargs(self, scriptname):
        if not self.request.config.getvalue("notoolsonpath"):
            # XXX we rely on script referring to the correct environment
            # we cannot use "(sys.executable,script)"
            # because on windows the script is e.g. a py.test.exe
            return (sys.executable, _pytest_fullpath,) # noqa
        else:
            pytest.skip("cannot run %r with --no-tools-on-path" % scriptname)

    def runpython(self, script, prepend=True):
        if prepend:
            s = self._getsysprepend()
            if s:
                script.write(s + "\n" + script.read())
        return self.run(sys.executable, script)

    def _getsysprepend(self):
        if self.request.config.getvalue("notoolsonpath"):
            s = "import sys;sys.path.insert(0,%r);" % str(py._pydir.dirpath())
        else:
            s = ""
        return s

    def runpython_c(self, command):
        command = self._getsysprepend() + command
        return self.run(sys.executable, "-c", command)

    def runpytest(self, *args):
        p = py.path.local.make_numbered_dir(prefix="runpytest-",
            keep=None, rootdir=self.tmpdir)
        args = ('--basetemp=%s' % p, ) + args
        #for x in args:
        #    if '--confcutdir' in str(x):
        #        break
        #else:
        #    pass
        #    args = ('--confcutdir=.',) + args
        plugins = [x for x in self.plugins if isinstance(x, str)]
        if plugins:
            args = ('-p', plugins[0]) + args
        return self.runpybin("py.test", *args)

    def spawn_pytest(self, string, expect_timeout=10.0):
        if self.request.config.getvalue("notoolsonpath"):
            pytest.skip("--no-tools-on-path prevents running pexpect-spawn tests")
        basetemp = self.tmpdir.mkdir("pexpect")
        invoke = " ".join(map(str, self._getpybinargs("py.test")))
        cmd = "%s --basetemp=%s %s" % (invoke, basetemp, string)
        return self.spawn(cmd, expect_timeout=expect_timeout)

    def spawn(self, cmd, expect_timeout=10.0):
        pexpect = pytest.importorskip("pexpect", "3.0")
        if hasattr(sys, 'pypy_version_info') and '64' in platform.machine():
            pytest.skip("pypy-64 bit not supported")
        if sys.platform == "darwin":
            pytest.xfail("pexpect does not work reliably on darwin?!")
        if sys.platform.startswith("freebsd"):
            pytest.xfail("pexpect does not work reliably on freebsd")
        logfile = self.tmpdir.join("spawn.out").open("wb")
        child = pexpect.spawn(cmd, logfile=logfile)
        self.request.addfinalizer(logfile.close)
        child.timeout = expect_timeout
        return child

def getdecoded(out):
        try:
            return out.decode("utf-8")
        except UnicodeDecodeError:
            return "INTERNAL not-utf8-decodeable, truncated string:\n%s" % (
                    py.io.saferepr(out),)

class ReportRecorder(object):
    def __init__(self, hook):
        self.hook = hook
        self.pluginmanager = hook._pm
        self.pluginmanager.register(self)

    def getcall(self, name):
        return self.hookrecorder.getcall(name)

    def popcall(self, name):
        return self.hookrecorder.popcall(name)

    def getcalls(self, names):
        """ return list of ParsedCall instances matching the given eventname. """
        return self.hookrecorder.getcalls(names)

    # functionality for test reports

    def getreports(self, names="pytest_runtest_logreport pytest_collectreport"):
        return [x.report for x in self.getcalls(names)]

    def matchreport(self, inamepart="",
        names="pytest_runtest_logreport pytest_collectreport", when=None):
        """ return a testreport whose dotted import path matches """
        l = []
        for rep in self.getreports(names=names):
            try:
                if not when and rep.when != "call" and rep.passed:
                    # setup/teardown passing reports - let's ignore those
                    continue
            except AttributeError:
                pass
            if when and getattr(rep, 'when', None) != when:
                continue
            if not inamepart or inamepart in rep.nodeid.split("::"):
                l.append(rep)
        if not l:
            raise ValueError("could not find test report matching %r: no test reports at all!" %
                (inamepart,))
        if len(l) > 1:
            raise ValueError("found more than one testreport matching %r: %s" %(
                             inamepart, l))
        return l[0]

    def getfailures(self, names='pytest_runtest_logreport pytest_collectreport'):
        return [rep for rep in self.getreports(names) if rep.failed]

    def getfailedcollections(self):
        return self.getfailures('pytest_collectreport')

    def listoutcomes(self):
        passed = []
        skipped = []
        failed = []
        for rep in self.getreports(
            "pytest_collectreport pytest_runtest_logreport"):
            if rep.passed:
                if getattr(rep, "when", None) == "call":
                    passed.append(rep)
            elif rep.skipped:
                skipped.append(rep)
            elif rep.failed:
                failed.append(rep)
        return passed, skipped, failed

    def countoutcomes(self):
        return [len(x) for x in self.listoutcomes()]

    def assertoutcome(self, passed=0, skipped=0, failed=0):
        realpassed, realskipped, realfailed = self.listoutcomes()
        assert passed == len(realpassed)
        assert skipped == len(realskipped)
        assert failed == len(realfailed)

    def clear(self):
        self.hookrecorder.calls[:] = []

    def unregister(self):
        self.pluginmanager.unregister(self)
        self.hookrecorder.finish_recording()

class LineComp:
    def __init__(self):
        self.stringio = py.io.TextIO()

    def assert_contains_lines(self, lines2):
        """ assert that lines2 are contained (linearly) in lines1.
            return a list of extralines found.
        """
        __tracebackhide__ = True
        val = self.stringio.getvalue()
        self.stringio.truncate(0)
        self.stringio.seek(0)
        lines1 = val.split("\n")
        return LineMatcher(lines1).fnmatch_lines(lines2)

class LineMatcher:
    def __init__(self,  lines):
        self.lines = lines

    def str(self):
        return "\n".join(self.lines)

    def _getlines(self, lines2):
        if isinstance(lines2, str):
            lines2 = py.code.Source(lines2)
        if isinstance(lines2, py.code.Source):
            lines2 = lines2.strip().lines
        return lines2

    def fnmatch_lines_random(self, lines2):
        lines2 = self._getlines(lines2)
        for line in lines2:
            for x in self.lines:
                if line == x or fnmatch(x, line):
                    print_("matched: ", repr(line))
                    break
            else:
                raise ValueError("line %r not found in output" % line)

    def get_lines_after(self, fnline):
        for i, line in enumerate(self.lines):
            if fnline == line or fnmatch(line, fnline):
                return self.lines[i+1:]
        raise ValueError("line %r not found in output" % fnline)

    def fnmatch_lines(self, lines2):
        def show(arg1, arg2):
            py.builtin.print_(arg1, arg2, file=sys.stderr)
        lines2 = self._getlines(lines2)
        lines1 = self.lines[:]
        nextline = None
        extralines = []
        __tracebackhide__ = True
        for line in lines2:
            nomatchprinted = False
            while lines1:
                nextline = lines1.pop(0)
                if line == nextline:
                    show("exact match:", repr(line))
                    break
                elif fnmatch(nextline, line):
                    show("fnmatch:", repr(line))
                    show("   with:", repr(nextline))
                    break
                else:
                    if not nomatchprinted:
                        show("nomatch:", repr(line))
                        nomatchprinted = True
                    show("    and:", repr(nextline))
                extralines.append(nextline)
            else:
                pytest.fail("remains unmatched: %r, see stderr" % (line,))