# Source code for _pytest.main

    """ core implementation of testing process: init, session, runtest loop. """
    import re

    import py
    import pytest, _pytest
    import os, sys, imp
    try:
        from collections import MutableMapping as MappingMixin
    except ImportError:
        from UserDict import DictMixin as MappingMixin

    from _pytest.runner import collect_one_node

    tracebackcutdir = py.path.local(_pytest.__file__).dirpath()

    # exitcodes for the command line
    EXIT_OK = 0
    EXIT_TESTSFAILED = 1
    EXIT_INTERRUPTED = 2
    EXIT_INTERNALERROR = 3
    EXIT_USAGEERROR = 4
    EXIT_NOTESTSCOLLECTED = 5

    name_re = re.compile("^[a-zA-Z_]\w*$")

    def pytest_addoption(parser):
        parser.addini("norecursedirs", "directory patterns to avoid for recursion",
            type="args", default=['.*', 'CVS', '_darcs', '{arch}', '*.egg'])
        parser.addini("testpaths", "directories to search for tests when no files or directories are given in the command line.",
            type="args", default=[])
        #parser.addini("dirpatterns",
        #    "patterns specifying possible locations of test files",
        #    type="linelist", default=["**/test_*.txt",
        #            "**/test_*.py", "**/*_test.py"]
        #)
        group = parser.getgroup("general", "running and selection options")
        group._addoption('-x', '--exitfirst', action="store_true", default=False,
                   dest="exitfirst",
                   help="exit instantly on first error or failed test."),
        group._addoption('--maxfail', metavar="num",
                   action="store", type=int, dest="maxfail", default=0,
                   help="exit after first num failures or errors.")
        group._addoption('--strict', action="store_true",
                   help="run pytest in strict mode, warnings become errors.")
        group._addoption("-c", metavar="file", type=str, dest="inifilename",
                   help="load configuration from `file` instead of trying to locate one of the implicit configuration files.")

        group = parser.getgroup("collect", "collection")
        group.addoption('--collectonly', '--collect-only', action="store_true",
            help="only collect tests, don't execute them."),
        group.addoption('--pyargs', action="store_true",
            help="try to interpret all arguments as python packages.")
        group.addoption("--ignore", action="append", metavar="path",
            help="ignore path during collection (multi-allowed).")
        # when changing this to --conf-cut-dir, config.py Conftest.setinitial
        # needs upgrading as well
        group.addoption('--confcutdir', dest="confcutdir", default=None,
            metavar="dir",
            help="only load conftest.py's relative to specified dir.")
        group.addoption('--noconftest', action="store_true",
            dest="noconftest", default=False,
            help="Don't load any conftest.py files.")

        group = parser.getgroup("debugconfig",
            "test session debugging and configuration")
        group.addoption('--basetemp', dest="basetemp", default=None, metavar="dir",
                   help="base temporary directory for this test run.")

    def pytest_namespace():
        collect = dict(Item=Item, Collector=Collector, File=File, Session=Session)
        return dict(collect=collect)

    def pytest_configure(config):
        pytest.config = config # compatibiltiy
        if config.option.exitfirst:
            config.option.maxfail = 1

    def wrap_session(config, doit):
        """Skeleton command line program"""
        session = Session(config)
        session.exitstatus = EXIT_OK
        initstate = 0
        try:
            try:
                config._do_configure()
                initstate = 1
                config.hook.pytest_sessionstart(session=session)
                initstate = 2
                session.exitstatus = doit(config, session) or 0
            except pytest.UsageError:
                raise
            except KeyboardInterrupt:
                excinfo = py.code.ExceptionInfo()
                config.hook.pytest_keyboard_interrupt(excinfo=excinfo)
                session.exitstatus = EXIT_INTERRUPTED
            except:
                excinfo = py.code.ExceptionInfo()
                config.notify_exception(excinfo, config.option)
                session.exitstatus = EXIT_INTERNALERROR
                if excinfo.errisinstance(SystemExit):
                    sys.stderr.write("mainloop: caught Spurious SystemExit!\n")

        finally:
            excinfo = None  # Explicitly break reference cycle.
            session.startdir.chdir()
            if initstate >= 2:
                config.hook.pytest_sessionfinish(
                    session=session,
                    exitstatus=session.exitstatus)
            config._ensure_unconfigure()
        return session.exitstatus

    def pytest_cmdline_main(config):
        return wrap_session(config, _main)

    def _main(config, session):
        """ default command line protocol for initialization, session,
        running tests and reporting. """
        config.hook.pytest_collection(session=session)
        config.hook.pytest_runtestloop(session=session)

        if session.testsfailed:
            return EXIT_TESTSFAILED
        elif session.testscollected == 0:
            return EXIT_NOTESTSCOLLECTED

    def pytest_collection(session):
        return session.perform_collect()

    def pytest_runtestloop(session):
        if session.config.option.collectonly:
            return True

        def getnextitem(i):
            # this is a function to avoid python2
            # keeping sys.exc_info set when calling into a test
            # python2 keeps sys.exc_info till the frame is left
            try:
                return session.items[i+1]
            except IndexError:
                return None

        for i, item in enumerate(session.items):
            nextitem = getnextitem(i)
            item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
            if session.shouldstop:
                raise session.Interrupted(session.shouldstop)
        return True

    def pytest_ignore_collect(path, config):
        p = path.dirpath()
        ignore_paths = config._getconftest_pathlist("collect_ignore", path=p)
        ignore_paths = ignore_paths or []
        excludeopt = config.getoption("ignore")
        if excludeopt:
            ignore_paths.extend([py.path.local(x) for x in excludeopt])
        return path in ignore_paths

    class FSHookProxy:
        def __init__(self, fspath, pm, remove_mods):
            self.fspath = fspath
            self.pm = pm
            self.remove_mods = remove_mods

        def __getattr__(self, name):
            x = self.pm.subset_hook_caller(name, remove_plugins=self.remove_mods)
            self.__dict__[name] = x
            return x

    def compatproperty(name):
        def fget(self):
            # deprecated - use pytest.name
            return getattr(pytest, name)

        return property(fget)

    class NodeKeywords(MappingMixin):
        def __init__(self, node):
            self.node = node
            self.parent = node.parent
            self._markers = {node.name: True}

        def __getitem__(self, key):
            try:
                return self._markers[key]
            except KeyError:
                if self.parent is None:
                    raise
                return self.parent.keywords[key]

        def __setitem__(self, key, value):
            self._markers[key] = value

        def __delitem__(self, key):
            raise ValueError("cannot delete key in keywords dict")

        def __iter__(self):
            seen = set(self._markers)
            if self.parent is not None:
                seen.update(self.parent.keywords)
            return iter(seen)

        def __len__(self):
            return len(self.__iter__())

        def keys(self):
            return list(self)

        def __repr__(self):
            return "<NodeKeywords for node %s>" % (self.node, )

    class Node(object):
        """ base class for Collector and Item the test collection tree.
        Collector subclasses have children, Items are terminal nodes."""

        def __init__(self, name, parent=None, config=None, session=None):
            #: a unique name within the scope of the parent node
            self.name = name

            #: the parent collector node.
            self.parent = parent

            #: the pytest config object
            self.config = config or parent.config

            #: the session this node is part of
            self.session = session or parent.session

            #: filesystem path where this node was collected from (can be None)
            self.fspath = getattr(parent, 'fspath', None)

            #: keywords/markers collected from all scopes
            self.keywords = NodeKeywords(self)

            #: allow adding of extra keywords to use for matching
            self.extra_keyword_matches = set()

            # used for storing artificial fixturedefs for direct parametrization
            self._name2pseudofixturedef = {}

        @property
        def ihook(self):
            """ fspath sensitive hook proxy used to call pytest hooks"""
            return self.session.gethookproxy(self.fspath)

        Module = compatproperty("Module")
        Class = compatproperty("Class")
        Instance = compatproperty("Instance")
        Function = compatproperty("Function")
        File = compatproperty("File")
        Item = compatproperty("Item")

        def _getcustomclass(self, name):
            cls = getattr(self, name)
            if cls != getattr(pytest, name):
                py.log._apiwarn("2.0", "use of node.%s is deprecated, "
                    "use pytest_pycollect_makeitem(...) to create custom "
                    "collection nodes" % name)
            return cls

        def __repr__(self):
            return "<%s %r>" %(self.__class__.__name__,
                               getattr(self, 'name', None))

        def warn(self, code, message):
            """ generate a warning with the given code and message for this
            item. """
            assert isinstance(code, str)
            fslocation = getattr(self, "location", None)
            if fslocation is None:
                fslocation = getattr(self, "fspath", None)
            else:
                fslocation = "%s:%s" % fslocation[:2]

            self.ihook.pytest_logwarning.call_historic(kwargs=dict(
                code=code, message=message,
                nodeid=self.nodeid, fslocation=fslocation))

        # methods for ordering nodes
        @property
        def nodeid(self):
            """ a ::-separated string denoting its collection tree address. """
            try:
                return self._nodeid
            except AttributeError:
                self._nodeid = x = self._makeid()
                return x

        def _makeid(self):
            return self.parent.nodeid + "::" + self.name

        def __hash__(self):
            return hash(self.nodeid)

        def setup(self):
            pass

        def teardown(self):
            pass

        def _memoizedcall(self, attrname, function):
            exattrname = "_ex_" + attrname
            failure = getattr(self, exattrname, None)
            if failure is not None:
                py.builtin._reraise(failure[0], failure[1], failure[2])
            if hasattr(self, attrname):
                return getattr(self, attrname)
            try:
                res = function()
            except py.builtin._sysex:
                raise
            except:
                failure = sys.exc_info()
                setattr(self, exattrname, failure)
                raise
            setattr(self, attrname, res)
            return res

        def listchain(self):
            """ return list of all parent collectors up to self,
                starting from root of collection tree. """
            chain = []
            item = self
            while item is not None:
                chain.append(item)
                item = item.parent
            chain.reverse()
            return chain

        def add_marker(self, marker):
            """ dynamically add a marker object to the node.

            ``marker`` can be a string or pytest.mark.* instance.
            """
            from _pytest.mark import MarkDecorator
            if isinstance(marker, py.builtin._basestring):
                marker = MarkDecorator(marker)
            elif not isinstance(marker, MarkDecorator):
                raise ValueError("is not a string or pytest.mark.* Marker")
            self.keywords[marker.name] = marker

        def get_marker(self, name):
            """ get a marker object from this node or None if
            the node doesn't have a marker with that name. """
            val = self.keywords.get(name, None)
            if val is not None:
                from _pytest.mark import MarkInfo, MarkDecorator
                if isinstance(val, (MarkDecorator, MarkInfo)):
                    return val

        def listextrakeywords(self):
            """ Return a set of all extra keywords in self and any parents."""
            extra_keywords = set()
            item = self
            for item in self.listchain():
                extra_keywords.update(item.extra_keyword_matches)
            return extra_keywords

        def listnames(self):
            return [x.name for x in self.listchain()]

        def addfinalizer(self, fin):
            """ register a function to be called when this node is finalized.

            This method can only be called when this node is active
            in a setup chain, for example during self.setup().
            """
            self.session._setupstate.addfinalizer(fin, self)

        def getparent(self, cls):
            """ get the next parent node (including ourself)
            which is an instance of the given class"""
            current = self
            while current and not isinstance(current, cls):
                current = current.parent
            return current

        def _prunetraceback(self, excinfo):
            pass

        def _repr_failure_py(self, excinfo, style=None):
            fm = self.session._fixturemanager
            if excinfo.errisinstance(fm.FixtureLookupError):
                return excinfo.value.formatrepr()
            tbfilter = True
            if self.config.option.fulltrace:
                style="long"
            else:
                self._prunetraceback(excinfo)
                tbfilter = False  # prunetraceback already does it
                if style == "auto":
                    style = "long"
            # XXX should excinfo.getrepr record all data and toterminal() process it?
            if style is None:
                if self.config.option.tbstyle == "short":
                    style = "short"
                else:
                    style = "long"

            return excinfo.getrepr(funcargs=True,
                                   showlocals=self.config.option.showlocals,
                                   style=style, tbfilter=tbfilter)

        repr_failure = _repr_failure_py

    class Collector(Node):
        """ Collector instances create children through collect()
            and thus iteratively build a tree.
        """

        class CollectError(Exception):
            """ an error during collection, contains a custom message. """

        def collect(self):
            """ returns a list of children (items and collectors)
                for this collection node.
            """
            raise NotImplementedError("abstract")

        def repr_failure(self, excinfo):
            """ represent a collection failure. """
            if excinfo.errisinstance(self.CollectError):
                exc = excinfo.value
                return str(exc.args[0])
            return self._repr_failure_py(excinfo, style="short")

        def _memocollect(self):
            """ internal helper method to cache results of calling collect(). """
            return self._memoizedcall('_collected', lambda: list(self.collect()))

        def _prunetraceback(self, excinfo):
            if hasattr(self, 'fspath'):
                traceback = excinfo.traceback
                ntraceback = traceback.cut(path=self.fspath)
                if ntraceback == traceback:
                    ntraceback = ntraceback.cut(excludepath=tracebackcutdir)
                excinfo.traceback = ntraceback.filter()

    class FSCollector(Collector):
        def __init__(self, fspath, parent=None, config=None, session=None):
            fspath = py.path.local(fspath) # xxx only for test_resultlog.py?
            name = fspath.basename
            if parent is not None:
                rel = fspath.relto(parent.fspath)
                if rel:
                    name = rel
                name = name.replace(os.sep, "/")
            super(FSCollector, self).__init__(name, parent, config, session)
            self.fspath = fspath

        def _makeid(self):
            relpath = self.fspath.relto(self.config.rootdir)
            if os.sep != "/":
                relpath = relpath.replace(os.sep, "/")
            return relpath

    class File(FSCollector):
        """ base class for collecting tests from a file. """

    class Item(Node):
        """ a basic test invocation item. Note that for a single function
        there might be multiple test invocation items.
        """
        nextitem = None

        def __init__(self, name, parent=None, config=None, session=None):
            super(Item, self).__init__(name, parent, config, session)
            self._report_sections = []

        def add_report_section(self, when, key, content):
            if content:
                self._report_sections.append((when, key, content))

        def reportinfo(self):
            return self.fspath, None, ""

        @property
        def location(self):
            try:
                return self._location
            except AttributeError:
                location = self.reportinfo()
                # bestrelpath is a quite slow function
                cache = self.config.__dict__.setdefault("_bestrelpathcache", {})
                try:
                    fspath = cache[location[0]]
                except KeyError:
                    fspath = self.session.fspath.bestrelpath(location[0])
                    cache[location[0]] = fspath
                location = (fspath, location[1], str(location[2]))
                self._location = location
                return location

    class NoMatch(Exception):
        """ raised if matching cannot locate a matching names. """

    class Interrupted(KeyboardInterrupt):
        """ signals an interrupted test run. """
        __module__ = 'builtins' # for py3

    class Session(FSCollector):
        Interrupted = Interrupted

        def __init__(self, config):
            FSCollector.__init__(self, config.rootdir, parent=None,
                                 config=config, session=self)
            self._fs2hookproxy = {}
            self.testsfailed = 0
            self.testscollected = 0
            self.shouldstop = False
            self.trace = config.trace.root.get("collection")
            self._norecursepatterns = config.getini("norecursedirs")
            self.startdir = py.path.local()
            self.config.pluginmanager.register(self, name="session")

        def _makeid(self):
            return ""

        @pytest.hookimpl(tryfirst=True)
        def pytest_collectstart(self):
            if self.shouldstop:
                raise self.Interrupted(self.shouldstop)

        @pytest.hookimpl(tryfirst=True)
        def pytest_runtest_logreport(self, report):
            if report.failed and not hasattr(report, 'wasxfail'):
                self.testsfailed += 1
                maxfail = self.config.getvalue("maxfail")
                if maxfail and self.testsfailed >= maxfail:
                    self.shouldstop = "stopping after %d failures" % (
                        self.testsfailed)
        pytest_collectreport = pytest_runtest_logreport

        def isinitpath(self, path):
            return path in self._initialpaths

        def gethookproxy(self, fspath):
            try:
                return self._fs2hookproxy[fspath]
            except KeyError:
                # check if we have the common case of running
                # hooks with all conftest.py filesall conftest.py
                pm = self.config.pluginmanager
                my_conftestmodules = pm._getconftestmodules(fspath)
                remove_mods = pm._conftest_plugins.difference(my_conftestmodules)
                if remove_mods:
                    # one or more conftests are not in use at this fspath
                    proxy = FSHookProxy(fspath, pm, remove_mods)
                else:
                    # all plugis are active for this fspath
                    proxy = self.config.hook

                self._fs2hookproxy[fspath] = proxy
                return proxy

        def perform_collect(self, args=None, genitems=True):
            hook = self.config.hook
            try:
                items = self._perform_collect(args, genitems)
                hook.pytest_collection_modifyitems(session=self,
                    config=self.config, items=items)
            finally:
                hook.pytest_collection_finish(session=self)
            self.testscollected = len(items)
            return items

        def _perform_collect(self, args, genitems):
            if args is None:
                args = self.config.args
            self.trace("perform_collect", self, args)
            self.trace.root.indent += 1
            self._notfound = []
            self._initialpaths = set()
            self._initialparts = []
            self.items = items = []
            for arg in args:
                parts = self._parsearg(arg)
                self._initialparts.append(parts)
                self._initialpaths.add(parts[0])
            rep = collect_one_node(self)
            self.ihook.pytest_collectreport(report=rep)
            self.trace.root.indent -= 1
            if self._notfound:
                errors = []
                for arg, exc in self._notfound:
                    line = "(no name %r in any of %r)" % (arg, exc.args[0])
                    errors.append("not found: %s\n%s" % (arg, line))
                    #XXX: test this
                raise pytest.UsageError(*errors)
            if not genitems:
                return rep.result
            else:
                if rep.passed:
                    for node in rep.result:
                        self.items.extend(self.genitems(node))
                return items

        def collect(self):
            for parts in self._initialparts:
                arg = "::".join(map(str, parts))
                self.trace("processing argument", arg)
                self.trace.root.indent += 1
                try:
                    for x in self._collect(arg):
                        yield x
                except NoMatch:
                    # we are inside a make_report hook so
                    # we cannot directly pass through the exception
                    self._notfound.append((arg, sys.exc_info()[1]))

                self.trace.root.indent -= 1

        def _collect(self, arg):
            names = self._parsearg(arg)
            path = names.pop(0)
            if path.check(dir=1):
                assert not names, "invalid arg %r" %(arg,)
                for path in path.visit(fil=lambda x: x.check(file=1),
                                       rec=self._recurse, bf=True, sort=True):
                    for x in self._collectfile(path):
                        yield x
            else:
                assert path.check(file=1)
                for x in self.matchnodes(self._collectfile(path), names):
                    yield x

        def _collectfile(self, path):
            ihook = self.gethookproxy(path)
            if not self.isinitpath(path):
                if ihook.pytest_ignore_collect(path=path, config=self.config):
                    return ()
            return ihook.pytest_collect_file(path=path, parent=self)

        def _recurse(self, path):
            ihook = self.gethookproxy(path.dirpath())
            if ihook.pytest_ignore_collect(path=path, config=self.config):
                return
            for pat in self._norecursepatterns:
                if path.check(fnmatch=pat):
                    return False
            ihook = self.gethookproxy(path)
            ihook.pytest_collect_directory(path=path, parent=self)
            return True

        def _tryconvertpyarg(self, x):
            mod = None
            path = [os.path.abspath('.')] + sys.path
            for name in x.split('.'):
                # ignore anything that's not a proper name here
                # else something like --pyargs will mess up '.'
                # since imp.find_module will actually sometimes work for it
                # but it's supposed to be considered a filesystem path
                # not a package
                if name_re.match(name) is None:
                    return x
                try:
                    fd, mod, type_ = imp.find_module(name, path)
                except ImportError:
                    return x
                else:
                    if fd is not None:
                        fd.close()

                if type_[2] != imp.PKG_DIRECTORY:
                    path = [os.path.dirname(mod)]
                else:
                    path = [mod]
            return mod

        def _parsearg(self, arg):
            """ return (fspath, names) tuple after checking the file exists. """
            arg = str(arg)
            if self.config.option.pyargs:
                arg = self._tryconvertpyarg(arg)
            parts = str(arg).split("::")
            relpath = parts[0].replace("/", os.sep)
            path = self.config.invocation_dir.join(relpath, abs=True)
            if not path.check():
                if self.config.option.pyargs:
                    msg = "file or package not found: "
                else:
                    msg = "file not found: "
                raise pytest.UsageError(msg + arg)
            parts[0] = path
            return parts

        def matchnodes(self, matching, names):
            self.trace("matchnodes", matching, names)
            self.trace.root.indent += 1
            nodes = self._matchnodes(matching, names)
            num = len(nodes)
            self.trace("matchnodes finished -> ", num, "nodes")
            self.trace.root.indent -= 1
            if num == 0:
                raise NoMatch(matching, names[:1])
            return nodes

        def _matchnodes(self, matching, names):
            if not matching or not names:
                return matching
            name = names[0]
            assert name
            nextnames = names[1:]
            resultnodes = []
            for node in matching:
                if isinstance(node, pytest.Item):
                    if not names:
                        resultnodes.append(node)
                    continue
                assert isinstance(node, pytest.Collector)
                rep = collect_one_node(node)
                if rep.passed:
                    has_matched = False
                    for x in rep.result:
                        if x.name == name:
                            resultnodes.extend(self.matchnodes([x], nextnames))
                            has_matched = True
                    # XXX accept IDs that don't have "()" for class instances
                    if not has_matched and len(rep.result) == 1 and x.name == "()":
                        nextnames.insert(0, name)
                        resultnodes.extend(self.matchnodes([x], nextnames))
                node.ihook.pytest_collectreport(report=rep)
            return resultnodes

        def genitems(self, node):
            self.trace("genitems", node)
            if isinstance(node, pytest.Item):
                node.ihook.pytest_itemcollected(item=node)
                yield node
            else:
                assert isinstance(node, pytest.Collector)
                rep = collect_one_node(node)
                if rep.passed:
                    for subnode in rep.result:
                        for x in self.genitems(subnode):
                            yield x
                node.ihook.pytest_collectreport(report=rep)