text:
"""Command line options, ini-file and conftest.py processing."""
import argparse
import collections.abc
import copy
import enum
import glob
import inspect
import os
import re
import shlex
import sys
import types
import warnings
from functools import lru_cache
from pathlib import Path
from textwrap import dedent
from types import FunctionType
from types import TracebackType
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import Generator
from typing import IO
from typing import Iterable
from typing import Iterator
from typing import List
from typing import Optional
from typing import Sequence
from typing import Set
from typing import TextIO
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import Union

import attr
from pluggy import HookimplMarker
from pluggy import HookspecMarker
from pluggy import PluginManager

import _pytest._code
import _pytest.deprecated
import _pytest.hookspec
from .exceptions import PrintHelp as PrintHelp
from .exceptions import UsageError as UsageError
from .findpaths import determine_setup
from _pytest._code import ExceptionInfo
from _pytest._code import filter_traceback
from _pytest._io import TerminalWriter
from _pytest.compat import final
from _pytest.compat import importlib_metadata
from _pytest.outcomes import fail
from _pytest.outcomes import Skipped
from _pytest.pathlib import absolutepath
from _pytest.pathlib import bestrelpath
from _pytest.pathlib import import_path
from _pytest.pathlib import ImportMode
from _pytest.pathlib import resolve_package_path
from _pytest.stash import Stash
from _pytest.warning_types import PytestConfigWarning
from _pytest.warning_types import warn_explicit_for

if TYPE_CHECKING:

    from _pytest._code.code import _TracebackStyle
    from _pytest.terminal import TerminalReporter
    from .argparsing import Argument

_PluggyPlugin = object
"""A type to represent plugin objects.

Plugins can be any namespace, so we can't narrow it down much, but we use an
alias to make the intent clear.

Ideally this type would be provided by pluggy itself.
"""

hookimpl = HookimplMarker("pytest")
hookspec = HookspecMarker("pytest")

@final
class ExitCode(enum.IntEnum):
    """Encodes the valid exit codes by pytest.

    Currently users and plugins may supply other exit codes as well.

    .. versionadded:: 5.0
    """

    #: Tests passed.
    OK = 0
    #: Tests failed.
    TESTS_FAILED = 1
    #: pytest was interrupted.
    INTERRUPTED = 2
    #: An internal error got in the way.
    INTERNAL_ERROR = 3
    #: pytest was misused.
    USAGE_ERROR = 4
    #: pytest couldn't find tests.
    NO_TESTS_COLLECTED = 5

class ConftestImportFailure(Exception):
    def __init__(
        self,
        path: Path,
        excinfo: Tuple[Type[Exception], Exception, TracebackType],
    ) -> None:
        super().__init__(path, excinfo)
        self.path = path
        self.excinfo = excinfo

    def __str__(self) -> str:
        return "{}: {} (from {})".format(
            self.excinfo[0].__name__, self.excinfo[1], self.path
        )

def filter_traceback_for_conftest_import_failure(
    entry: _pytest._code.TracebackEntry,
) -> bool:
    """Filter tracebacks entries which point to pytest internals or importlib.

    Make a special case for importlib because we use it to import test modules and conftest files
    in _pytest.pathlib.import_path.
    """
    return filter_traceback(entry) and "importlib" not in str(entry.path).split(os.sep)

def main(
    args: Optional[Union[List[str], "os.PathLike[str]"]] = None,
    plugins: Optional[Sequence[Union[str, _PluggyPlugin]]] = None,
) -> Union[int, ExitCode]:
    """Perform an in-process test run.

    :param args: List of command line arguments.
    :param plugins: List of plugin objects to be auto-registered during initialization.

    :returns: An exit code.
    """
    try:
        try:
            config = _prepareconfig(args, plugins)
        except ConftestImportFailure as e:
            exc_info = ExceptionInfo.from_exc_info(e.excinfo)
            tw = TerminalWriter(sys.stderr)
            tw.line(f"ImportError while loading conftest '{e.path}'.", red=True)
            exc_info.traceback = exc_info.traceback.filter(
                filter_traceback_for_conftest_import_failure
            )
            exc_repr = (
                exc_info.getrepr(style="short", chain=False)
                if exc_info.traceback
                else exc_info.exconly()
            )
            formatted_tb = str(exc_repr)
            for line in formatted_tb.splitlines():
                tw.line(line.rstrip(), red=True)
            return ExitCode.USAGE_ERROR
        else:
            try:
                ret: Union[ExitCode, int] = config.hook.pytest_cmdline_main(
                    config=config
                )
                try:
                    return ExitCode(ret)
                except ValueError:
                    return ret
            finally:
                config._ensure_unconfigure()
    except UsageError as e:
        tw = TerminalWriter(sys.stderr)
        for msg in e.args:
            tw.line(f"ERROR: {msg}\n", red=True)
        return ExitCode.USAGE_ERROR

def console_main() -> int:
    """The CLI entry point of pytest.

    This function is not meant for programmable use; use `main()` instead.
    """
    # https://docs.python.org/3/library/signal.html#note-on-sigpipe
    try:
        code = main()
        sys.stdout.flush()
        return code
    except BrokenPipeError:
        # Python flushes standard streams on exit; redirect remaining output
        # to devnull to avoid another BrokenPipeError at shutdown
        devnull = os.open(os.devnull, os.O_WRONLY)
        os.dup2(devnull, sys.stdout.fileno())
        return 1  # Python exits with error code 1 on EPIPE

class cmdline:  # compatibility namespace
    main = staticmethod(main)

def filename_arg(path: str, optname: str) -> str:
    """Argparse type validator for filename arguments.

    :path: Path of filename.
    :optname: Name of the option.
    """
    if os.path.isdir(path):
        raise UsageError(f"{optname} must be a filename, given: {path}")
    return path

def directory_arg(path: str, optname: str) -> str:
    """Argparse type validator for directory arguments.

    :path: Path of directory.
    :optname: Name of the option.
    """
    if not os.path.isdir(path):
        raise UsageError(f"{optname} must be a directory, given: {path}")
    return path

# Plugins that cannot be disabled via "-p no:X" currently.
essential_plugins = (
    "mark",
    "main",
    "runner",
    "fixtures",
    "helpconfig",  # Provides -p.
)

default_plugins = essential_plugins + (
    "python",
    "terminal",
    "debugging",
    "unittest",
    "capture",
    "skipping",
    "legacypath",
    "tmpdir",
    "monkeypatch",
    "recwarn",
    "pastebin",
    "nose",
    "assertion",
    "junitxml",
    "doctest",
    "cacheprovider",
    "freeze_support",
    "setuponly",
    "setupplan",
    "stepwise",
    "warnings",
    "logging",
    "reports",
    "python_path",
    *(["unraisableexception", "threadexception"] if sys.version_info >= (3, 8) else []),
    "faulthandler",
)

builtin_plugins = set(default_plugins)
builtin_plugins.add("pytester")
builtin_plugins.add("pytester_assertions")

def get_config(
    args: Optional[List[str]] = None,
    plugins: Optional[Sequence[Union[str, _PluggyPlugin]]] = None,
) -> "Config":
    # subsequent calls to main will create a fresh instance
    pluginmanager = PytestPluginManager()
    config = Config(
        pluginmanager,
        invocation_params=Config.InvocationParams(
            args=args or (),
            plugins=plugins,
            dir=Path.cwd(),
        ),
    )

    if args is not None:
        # Handle any "-p no:plugin" args.
        pluginmanager.consider_preparse(args, exclude_only=True)

    for spec in default_plugins:
        pluginmanager.import_plugin(spec)

    return config

def get_plugin_manager() -> "PytestPluginManager":
    """Obtain a new instance of the
    :py:class:`pytest.PytestPluginManager`, with default plugins
    already loaded.

    This function can be used by integration with other tools, like hooking
    into pytest to run tests into an IDE.
    """
    return get_config().pluginmanager

def _prepareconfig(
    args: Optional[Union[List[str], "os.PathLike[str]"]] = None,
    plugins: Optional[Sequence[Union[str, _PluggyPlugin]]] = None,
) -> "Config":
    if args is None:
        args = sys.argv[1:]
    elif isinstance(args, os.PathLike):
        args = [os.fspath(args)]
    elif not isinstance(args, list):
        msg = (  # type:ignore[unreachable]
            "`args` parameter expected to be a list of strings, got: {!r} (type: {})"
        )
        raise TypeError(msg.format(args, type(args)))

    config = get_config(args, plugins)
    pluginmanager = config.pluginmanager
    try:
        if plugins:
            for plugin in plugins:
                if isinstance(plugin, str):
                    pluginmanager.consider_pluginarg(plugin)
                else:
                    pluginmanager.register(plugin)
        config = pluginmanager.hook.pytest_cmdline_parse(
            pluginmanager=pluginmanager, args=args
        )
        return config
    except BaseException:
        config._ensure_unconfigure()
        raise

def _get_directory(path: Path) -> Path:
    """Get the directory of a path - itself if already a directory."""
    if path.is_file():
        return path.parent
    else:
        return path

def _get_legacy_hook_marks(
    method: Any,
    hook_type: str,
    opt_names: Tuple[str, ...],
) -> Dict[str, bool]:
    if TYPE_CHECKING:
        # abuse typeguard from importlib to avoid massive method type union thats lacking a alias
        assert inspect.isroutine(method)
    known_marks: set[str] = {m.name for m in getattr(method, "pytestmark", [])}
    must_warn: list[str] = []
    opts: dict[str, bool] = {}
    for opt_name in opt_names:
        opt_attr = getattr(method, opt_name, AttributeError)
        if opt_attr is not AttributeError:
            must_warn.append(f"{opt_name}={opt_attr}")
            opts[opt_name] = True
        elif opt_name in known_marks:
            must_warn.append(f"{opt_name}=True")
            opts[opt_name] = True
        else:
            opts[opt_name] = False
    if must_warn:
        hook_opts = ", ".join(must_warn)
        message = _pytest.deprecated.HOOK_LEGACY_MARKING.format(
            type=hook_type,
            fullname=method.__qualname__,
            hook_opts=hook_opts,
        )
        warn_explicit_for(cast(FunctionType, method), message)
    return opts

@final
class PytestPluginManager(PluginManager):
    """A :py:class:`pluggy.PluginManager <pluggy.PluginManager>` with
    additional pytest-specific functionality:

    * Loading plugins from the command line, ``PYTEST_PLUGINS`` env variable and
      ``pytest_plugins`` global variables found in plugins being loaded.
    * ``conftest.py`` loading during start-up.
    """

    def __init__(self) -> None:
        import _pytest.assertion

        super().__init__("pytest")

        # -- State related to local conftest plugins.
        # All loaded conftest modules.
        self._conftest_plugins: Set[types.ModuleType] = set()
        # All conftest modules applicable for a directory.
        # This includes the directory's own conftest modules as well
        # as those of its parent directories.
        self._dirpath2confmods: Dict[Path, List[types.ModuleType]] = {}
        # Cutoff directory above which conftests are no longer discovered.
        self._confcutdir: Optional[Path] = None
        # If set, conftest loading is skipped.
        self._noconftest = False

        # _getconftestmodules()'s call to _get_directory() causes a stat
        # storm when it's called potentially thousands of times in a test
        # session (#9478), often with the same path, so cache it.
        self._get_directory = lru_cache(256)(_get_directory)

        self._duplicatepaths: Set[Path] = set()

        # plugins that were explicitly skipped with pytest.skip
        # list of (module name, skip reason)
        # previously we would issue a warning when a plugin was skipped, but
        # since we refactored warnings as first citizens of Config, they are
        # just stored here to be used later.
        self.skipped_plugins: List[Tuple[str, str]] = []

        self.add_hookspecs(_pytest.hookspec)
        self.register(self)
        if os.environ.get("PYTEST_DEBUG"):
            err: IO[str] = sys.stderr
            encoding: str = getattr(err, "encoding", "utf8")
            try:
                err = open(
                    os.dup(err.fileno()),
                    mode=err.mode,
                    buffering=1,
                    encoding=encoding,
                )
            except Exception:
                pass
            self.trace.root.setwriter(err.write)
            self.enable_tracing()

        # Config._consider_importhook will set a real object if required.
        self.rewrite_hook = _pytest.assertion.DummyRewriteHook()
        # Used to know when we are importing conftests after the pytest_configure stage.
        self._configured = False

    def parse_hookimpl_opts(self, plugin: _PluggyPlugin, name: str):
        # pytest hooks are always prefixed with "pytest_",
        # so we avoid accessing possibly non-readable attributes
        # (see issue #1073).
        if not name.startswith("pytest_"):
            return
        # Ignore names which can not be hooks.
        if name == "pytest_plugins":
            return

        opts = super().parse_hookimpl_opts(plugin, name)
        if opts is not None:
            return opts

        method = getattr(plugin, name)
        # Consider only actual functions for hooks (#3775).
        if not inspect.isroutine(method):
            return
        # Collect unmarked hooks as long as they have the `pytest_' prefix.
        return _get_legacy_hook_marks(
            method, "impl", ("tryfirst", "trylast", "optionalhook", "hookwrapper")
        )

    def parse_hookspec_opts(self, module_or_class, name: str):
        opts = super().parse_hookspec_opts(module_or_class, name)
        if opts is None:
            method = getattr(module_or_class, name)
            if name.startswith("pytest_"):
                opts = _get_legacy_hook_marks(
                    method,
                    "spec",
                    ("firstresult", "historic"),
                )
        return opts

    def register(
        self, plugin: _PluggyPlugin, name: Optional[str] = None
    ) -> Optional[str]:
        if name in _pytest.deprecated.DEPRECATED_EXTERNAL_PLUGINS:
            warnings.warn(
                PytestConfigWarning(
                    "{} plugin has been merged into the core, "
                    "please remove it from your requirements.".format(
                        name.replace("_", "-")
                    )
                )
            )
            return None
        ret: Optional[str] = super().register(plugin, name)
        if ret:
            self.hook.pytest_plugin_registered.call_historic(
                kwargs=dict(plugin=plugin, manager=self)
            )

            if isinstance(plugin, types.ModuleType):
                self.consider_module(plugin)
        return ret

    def getplugin(self, name: str):
        # Support deprecated naming because plugins (xdist e.g.) use it.
        plugin: Optional[_PluggyPlugin] = self.get_plugin(name)
        return plugin

    def hasplugin(self, name: str) -> bool:
        """Return whether a plugin with the given name is registered."""
        return bool(self.get_plugin(name))

    def pytest_configure(self, config: "Config") -> None:
        """:meta private:"""
        # XXX now that the pluginmanager exposes hookimpl(tryfirst...)
        # we should remove tryfirst/trylast as markers.
        config.addinivalue_line(
            "markers",
            "tryfirst: mark a hook implementation function such that the "
            "plugin machinery will try to call it first/as early as possible. "
            "DEPRECATED, use @pytest.hookimpl(tryfirst=True) instead.",
        )
        config.addinivalue_line(
            "markers",
            "trylast: mark a hook implementation function such that the "
            "plugin machinery will try to call it last/as late as possible. "
            "DEPRECATED, use @pytest.hookimpl(trylast=True) instead.",
        )
        self._configured = True

    #
    # Internal API for local conftest plugin handling.
    #
    def _set_initial_conftests(
        self, namespace: argparse.Namespace, rootpath: Path
    ) -> None:
        """Load initial conftest files given a preparsed "namespace".

        As conftest files may add their own command line options which have
        arguments ('--my-opt somepath') we might get some false positives.
        All builtin and 3rd party plugins will have been loaded, however, so
        common options will not confuse our logic here.
        """
        current = Path.cwd()
        self._confcutdir = (
            absolutepath(current / namespace.confcutdir)
            if namespace.confcutdir
            else None
        )
        self._noconftest = namespace.noconftest
        self._using_pyargs = namespace.pyargs
        testpaths = namespace.file_or_dir
        foundanchor = False
        for testpath in testpaths:
            path = str(testpath)
            # remove node-id syntax
            i = path.find("::")
            if i != -1:
                path = path[:i]
            anchor = absolutepath(current / path)
            if anchor.exists():  # we found some file object
                self._try_load_conftest(anchor, namespace.importmode, rootpath)
                foundanchor = True
        if not foundanchor:
            self._try_load_conftest(current, namespace.importmode, rootpath)

    def _is_in_confcutdir(self, path: Path) -> bool:
        """Whether a path is within the confcutdir.

        When false, should not load conftest.
        """
        if self._confcutdir is None:
            return True
        return path not in self._confcutdir.parents

    def _try_load_conftest(
        self, anchor: Path, importmode: Union[str, ImportMode], rootpath: Path
    ) -> None:
        self._getconftestmodules(anchor, importmode, rootpath)
        # let's also consider test* subdirs
        if anchor.is_dir():
            for x in anchor.glob("test*"):
                if x.is_dir():
                    self._getconftestmodules(x, importmode, rootpath)

    def _getconftestmodules(
        self, path: Path, importmode: Union[str, ImportMode], rootpath: Path
    ) -> Sequence[types.ModuleType]:
        if self._noconftest:
            return []

        directory = self._get_directory(path)

        # Optimization: avoid repeated searches in the same directory.
        # Assumes always called with same importmode and rootpath.
        existing_clist = self._dirpath2confmods.get(directory)
        if existing_clist is not None:
            return existing_clist

        # XXX these days we may rather want to use config.rootpath
        # and allow users to opt into looking into the rootdir parent
        # directories instead of requiring to specify confcutdir.
        clist = []
        for parent in reversed((directory, *directory.parents)):
            if self._is_in_confcutdir(parent):
                conftestpath = parent / "conftest.py"
                if conftestpath.is_file():
                    mod = self._importconftest(conftestpath, importmode, rootpath)
                    clist.append(mod)
        self._dirpath2confmods[directory] = clist
        return clist

    def _rget_with_confmod(
        self,
        name: str,
        path: Path,
        importmode: Union[str, ImportMode],
        rootpath: Path,
    ) -> Tuple[types.ModuleType, Any]:
        modules = self._getconftestmodules(path, importmode, rootpath=rootpath)
        for mod in reversed(modules):
            try:
                return mod, getattr(mod, name)
            except AttributeError:
                continue
        raise KeyError(name)

    def _importconftest(
        self, conftestpath: Path, importmode: Union[str, ImportMode], rootpath: Path
    ) -> types.ModuleType:
        existing = self.get_plugin(str(conftestpath))
        if existing is not None:
            return cast(types.ModuleType, existing)

        pkgpath = resolve_package_path(conftestpath)
        if pkgpath is None:
            _ensure_removed_sysmodule(conftestpath.stem)

        try:
            mod = import_path(conftestpath, mode=importmode, root=rootpath)
        except Exception as e:
            assert e.__traceback__ is not None
            exc_info = (type(e), e, e.__traceback__)
            raise ConftestImportFailure(conftestpath, exc_info) from e

        self._check_non_top_pytest_plugins(mod, conftestpath)

        self._conftest_plugins.add(mod)
        dirpath = conftestpath.parent
        if dirpath in self._dirpath2confmods:
            for path, mods in self._dirpath2confmods.items():
                if dirpath in path.parents or path == dirpath:
                    assert mod not in mods
                    mods.append(mod)
        self.trace(f"loading conftestmodule {mod!r}")
        self.consider_conftest(mod)
        return mod

    def _check_non_top_pytest_plugins(
        self,
        mod: types.ModuleType,
        conftestpath: Path,
    ) -> None:
        if (
            hasattr(mod, "pytest_plugins")
            and self._configured
            and not self._using_pyargs
        ):
            msg = (
                "Defining 'pytest_plugins' in a non-top-level conftest is no longer supported:\n"
                "It affects the entire test suite instead of just below the conftest as expected.\n"
                "  {}\n"
                "Please move it to a top level conftest file at the rootdir:\n"
                "  {}\n"
                "For more information, visit:\n"
                "  https://docs.pytest.org/en/stable/deprecations.html#pytest-plugins-in-non-top-level-conftest-files"
            )
            fail(msg.format(conftestpath, self._confcutdir), pytrace=False)

    #
    # API for bootstrapping plugin loading
    #
    #

    def consider_preparse(
        self, args: Sequence[str], *, exclude_only: bool = False
    ) -> None:
        """:meta private:"""
        i = 0
        n = len(args)
        while i < n:
            opt = args[i]
            i += 1
            if isinstance(opt, str):
                if opt == "-p":
                    try:
                        parg = args[i]
                    except IndexError:
                        return
                    i += 1
                elif opt.startswith("-p"):
                    parg = opt[2:]
                else:
                    continue
                if exclude_only and not parg.startswith("no:"):
                    continue
                self.consider_pluginarg(parg)

    def consider_pluginarg(self, arg: str) -> None:
        """:meta private:"""
        if arg.startswith("no:"):
            name = arg[3:]
            if name in essential_plugins:
                raise UsageError("plugin %s cannot be disabled" % name)

            # PR #4304: remove stepwise if cacheprovider is blocked.
            if name == "cacheprovider":
                self.set_blocked("stepwise")
                self.set_blocked("pytest_stepwise")

            self.set_blocked(name)
            if not name.startswith("pytest_"):
                self.set_blocked("pytest_" + name)
        else:
            name = arg
            # Unblock the plugin.  None indicates that it has been blocked.
            # There is no interface with pluggy for this.
            if self._name2plugin.get(name, -1) is None:
                del self._name2plugin[name]
            if not name.startswith("pytest_"):
                if self._name2plugin.get("pytest_" + name, -1) is None:
                    del self._name2plugin["pytest_" + name]
            self.import_plugin(arg, consider_entry_points=True)

    def consider_conftest(self, conftestmodule: types.ModuleType) -> None:
        """:meta private:"""
        self.register(conftestmodule, name=conftestmodule.__file__)

    def consider_env(self) -> None:
        """:meta private:"""
        self._import_plugin_specs(os.environ.get("PYTEST_PLUGINS"))

    def consider_module(self, mod: types.ModuleType) -> None:
        """:meta private:"""
        self._import_plugin_specs(getattr(mod, "pytest_plugins", []))

    def _import_plugin_specs(
        self, spec: Union[None, types.ModuleType, str, Sequence[str]]
    ) -> None:
        plugins = _get_plugin_specs_as_list(spec)
        for import_spec in plugins:
            self.import_plugin(import_spec)

    def import_plugin(self, modname: str, consider_entry_points: bool = False) -> None:
        """Import a plugin with ``modname``.

        If ``consider_entry_points`` is True, entry point names are also
        considered to find a plugin.
        """
        # Most often modname refers to builtin modules, e.g. "pytester",
        # "terminal" or "capture".  Those plugins are registered under their
        # basename for historic purposes but must be imported with the
        # _pytest prefix.
        assert isinstance(modname, str), (
            "module name as text required, got %r" % modname
        )
        if self.is_blocked(modname) or self.get_plugin(modname) is not None:
            return

        importspec = "_pytest." + modname if modname in builtin_plugins else modname
        self.rewrite_hook.mark_rewrite(importspec)

        if consider_entry_points:
            loaded = self.load_setuptools_entrypoints("pytest11", name=modname)
            if loaded:
                return

        try:
            __import__(importspec)
        except ImportError as e:
            raise ImportError(
                f'Error importing plugin "{modname}": {e.args[0]}'
            ).with_traceback(e.__traceback__) from e

        except Skipped as e:
            self.skipped_plugins.append((modname, e.msg or ""))
        else:
            mod = sys.modules[importspec]
            self.register(mod, modname)

def _get_plugin_specs_as_list(
    specs: Union[None, types.ModuleType, str, Sequence[str]]
) -> List[str]:
    """Parse a plugins specification into a list of plugin names."""
    # None means empty.
    if specs is None:
        return []
    # Workaround for #3899 - a submodule which happens to be called "pytest_plugins".
    if isinstance(specs, types.ModuleType):
        return []
    # Comma-separated list.
    if isinstance(specs, str):
        return specs.split(",") if specs else []
    # Direct specification.
    if isinstance(specs, collections.abc.Sequence):
        return list(specs)
    raise UsageError(
        "Plugins may be specified as a sequence or a ','-separated string of plugin names. Got: %r"
        % specs
    )

def _ensure_removed_sysmodule(modname: str) -> None:
    try:
        del sys.modules[modname]
    except KeyError:
        pass

class Notset:
    def __repr__(self):
        return "<NOTSET>"

notset = Notset()

def _iter_rewritable_modules(package_files: Iterable[str]) -> Iterator[str]:
    """Given an iterable of file names in a source distribution, return the "names" that should
    be marked for assertion rewrite.

    For example the package "pytest_mock/__init__.py" should be added as "pytest_mock" in
    the assertion rewrite mechanism.

    This function has to deal with dist-info based distributions and egg based distributions
    (which are still very much in use for "editable" installs).

    Here are the file names as seen in a dist-info based distribution:

        pytest_mock/__init__.py
        pytest_mock/_version.py
        pytest_mock/plugin.py
        pytest_mock.egg-info/PKG-INFO

    Here are the file names as seen in an egg based distribution:

        src/pytest_mock/__init__.py
        src/pytest_mock/_version.py
        src/pytest_mock/plugin.py
        src/pytest_mock.egg-info/PKG-INFO
        LICENSE
        setup.py

    We have to take in account those two distribution flavors in order to determine which
    names should be considered for assertion rewriting.

    More information:
        https://github.com/pytest-dev/pytest-mock/issues/167
    """
    package_files = list(package_files)
    seen_some = False
    for fn in package_files:
        is_simple_module = "/" not in fn and fn.endswith(".py")
        is_package = fn.count("/") == 1 and fn.endswith("__init__.py")
        if is_simple_module:
            module_name, _ = os.path.splitext(fn)
            # we ignore "setup.py" at the root of the distribution
            # as well as editable installation finder modules made by setuptools
            if module_name != "setup" and not module_name.startswith("__editable__"):
                seen_some = True
                yield module_name
        elif is_package:
            package_name = os.path.dirname(fn)
            seen_some = True
            yield package_name

    if not seen_some:
        # At this point we did not find any packages or modules suitable for assertion
        # rewriting, so we try again by stripping the first path component (to account for
        # "src" based source trees for example).
        # This approach lets us have the common case continue to be fast, as egg-distributions
        # are rarer.
        new_package_files = []
        for fn in package_files:
            parts = fn.split("/")
            new_fn = "/".join(parts[1:])
            if new_fn:
                new_package_files.append(new_fn)
        if new_package_files:
            yield from _iter_rewritable_modules(new_package_files)

def _args_converter(args: Iterable[str]) -> Tuple[str, ...]:
    return tuple(args)

@final
class Config:
    """Access to configuration values, pluginmanager and plugin hooks.

    :param PytestPluginManager pluginmanager:
        A pytest PluginManager.

    :param InvocationParams invocation_params:
        Object containing parameters regarding the :func:`pytest.main`
        invocation.
    """

    @final
    @attr.s(frozen=True, auto_attribs=True)
    class InvocationParams:
        """Holds parameters passed during :func:`pytest.main`.

        The object attributes are read-only.

        .. versionadded:: 5.1

        .. note::

            Note that the environment variable ``PYTEST_ADDOPTS`` and the ``addopts``
            ini option are handled by pytest, not being included in the ``args`` attribute.

            Plugins accessing ``InvocationParams`` must be aware of that.
        """

        args: Tuple[str, ...] = attr.ib(converter=_args_converter)
        """The command-line arguments as passed to :func:`pytest.main`."""
        plugins: Optional[Sequence[Union[str, _PluggyPlugin]]]
        """Extra plugins, might be `None`."""
        dir: Path
        """The directory from which :func:`pytest.main` was invoked."""

    class ArgsSource(enum.Enum):
        """Indicates the source of the test arguments.

        .. versionadded:: 7.2
        """

        #: Command line arguments.
        ARGS = enum.auto()
        #: Invocation directory.
        INCOVATION_DIR = enum.auto()
        #: 'testpaths' configuration value.
        TESTPATHS = enum.auto()

    def __init__(
        self,
        pluginmanager: PytestPluginManager,
        *,
        invocation_params: Optional[InvocationParams] = None,
    ) -> None:
        from .argparsing import Parser, FILE_OR_DIR

        if invocation_params is None:
            invocation_params = self.InvocationParams(
                args=(), plugins=None, dir=Path.cwd()
            )

        self.option = argparse.Namespace()
        """Access to command line option as attributes.

        :type: argparse.Namespace
        """

        self.invocation_params = invocation_params
        """The parameters with which pytest was invoked.

        :type: InvocationParams
        """

        _a = FILE_OR_DIR
        self._parser = Parser(
            usage=f"%(prog)s [options] [{_a}] [{_a}] [...]",
            processopt=self._processopt,
            _ispytest=True,
        )
        self.pluginmanager = pluginmanager
        """The plugin manager handles plugin registration and hook invocation.

        :type: PytestPluginManager
        """

        self.stash = Stash()
        """A place where plugins can store information on the config for their
        own use.

        :type: Stash
        """
        # Deprecated alias. Was never public. Can be removed in a few releases.
        self._store = self.stash

        from .compat import PathAwareHookProxy

        self.trace = self.pluginmanager.trace.root.get("config")
        self.hook = PathAwareHookProxy(self.pluginmanager.hook)
        self._inicache: Dict[str, Any] = {}
        self._override_ini: Sequence[str] = ()
        self._opt2dest: Dict[str, str] = {}
        self._cleanup: List[Callable[[], None]] = []
        self.pluginmanager.register(self, "pytestconfig")
        self._configured = False
        self.hook.pytest_addoption.call_historic(
            kwargs=dict(parser=self._parser, pluginmanager=self.pluginmanager)
        )

        if TYPE_CHECKING:
            from _pytest.cacheprovider import Cache

            self.cache: Optional[Cache] = None

    @property
    def rootpath(self) -> Path:
        """The path to the :ref:`rootdir <rootdir>`.

        :type: pathlib.Path

        .. versionadded:: 6.1
        """
        return self._rootpath

    @property
    def inipath(self) -> Optional[Path]:
        """The path to the :ref:`configfile <configfiles>`.

        :type: Optional[pathlib.Path]

        .. versionadded:: 6.1
        """
        return self._inipath

    def add_cleanup(self, func: Callable[[], None]) -> None:
        """Add a function to be called when the config object gets out of
        use (usually coinciding with pytest_unconfigure)."""
        self._cleanup.append(func)

    def _do_configure(self) -> None:
        assert not self._configured
        self._configured = True
        with warnings.catch_warnings():
            warnings.simplefilter("default")
            self.hook.pytest_configure.call_historic(kwargs=dict(config=self))

    def _ensure_unconfigure(self) -> None:
        if self._configured:
            self._configured = False
            self.hook.pytest_unconfigure(config=self)
            self.hook.pytest_configure._call_history = []
        while self._cleanup:
            fin = self._cleanup.pop()
            fin()

    def get_terminal_writer(self) -> TerminalWriter:
        terminalreporter: TerminalReporter = self.pluginmanager.get_plugin(
            "terminalreporter"
        )
        return terminalreporter._tw

    def pytest_cmdline_parse(
        self, pluginmanager: PytestPluginManager, args: List[str]
    ) -> "Config":
        try:
            self.parse(args)
        except UsageError:

            # Handle --version and --help here in a minimal fashion.
            # This gets done via helpconfig normally, but its
            # pytest_cmdline_main is not called in case of errors.
            if getattr(self.option, "version", False) or "--version" in args:
                from _pytest.helpconfig import showversion

                showversion(self)
            elif (
                getattr(self.option, "help", False) or "--help" in args or "-h" in args
            ):
                self._parser._getparser().print_help()
                sys.stdout.write(
                    "\nNOTE: displaying only minimal help due to UsageError.\n\n"
                )

            raise

        return self

    def notify_exception(
        self,
        excinfo: ExceptionInfo[BaseException],
        option: Optional[argparse.Namespace] = None,
    ) -> None:
        if option and getattr(option, "fulltrace", False):
            style: _TracebackStyle = "long"
        else:
            style = "native"
        excrepr = excinfo.getrepr(
            funcargs=True, showlocals=getattr(option, "showlocals", False), style=style
        )
        res = self.hook.pytest_internalerror(excrepr=excrepr, excinfo=excinfo)
        if not any(res):
            for line in str(excrepr).split("\n"):
                sys.stderr.write("INTERNALERROR> %s\n" % line)
                sys.stderr.flush()

    def cwd_relative_nodeid(self, nodeid: str) -> str:
        # nodeid's are relative to the rootpath, compute relative to cwd.
        if self.invocation_params.dir != self.rootpath:
            fullpath = self.rootpath / nodeid
            nodeid = bestrelpath(self.invocation_params.dir, fullpath)
        return nodeid

    @classmethod
    def fromdictargs(cls, option_dict, args) -> "Config":
        """Constructor usable for subprocesses."""
        config = get_config(args)
        config.option.__dict__.update(option_dict)
        config.parse(args, addopts=False)
        for x in config.option.plugins:
            config.pluginmanager.consider_pluginarg(x)
        return config

    def _processopt(self, opt: "Argument") -> None:
        for name in opt._short_opts + opt._long_opts:
            self._opt2dest[name] = opt.dest

        if hasattr(opt, "default"):
            if not hasattr(self.option, opt.dest):
                setattr(self.option, opt.dest, opt.default)

    @hookimpl(trylast=True)
    def pytest_load_initial_conftests(self, early_config: "Config") -> None:
        self.pluginmanager._set_initial_conftests(
            early_config.known_args_namespace, rootpath=early_config.rootpath
        )

    def _initini(self, args: Sequence[str]) -> None:
        ns, unknown_args = self._parser.parse_known_and_unknown_args(
            args, namespace=copy.copy(self.option)
        )
        rootpath, inipath, inicfg = determine_setup(
            ns.inifilename,
            ns.file_or_dir + unknown_args,
            rootdir_cmd_arg=ns.rootdir or None,
            config=self,
        )
        self._rootpath = rootpath
        self._inipath = inipath
        self.inicfg = inicfg
        self._parser.extra_info["rootdir"] = str(self.rootpath)
        self._parser.extra_info["inifile"] = str(self.inipath)
        self._parser.addini("addopts", "Extra command line options", "args")
        self._parser.addini("minversion", "Minimally required pytest version")
        self._parser.addini(
            "required_plugins",
            "Plugins that must be present for pytest to run",
            type="args",
            default=[],
        )
        self._override_ini = ns.override_ini or ()

    def _consider_importhook(self, args: Sequence[str]) -> None:
        """Install the PEP 302 import hook if using assertion rewriting.

        Needs to parse the --assert=<mode> option from the commandline
        and find all the installed plugins to mark them for rewriting
        by the importhook.
        """
        ns, unknown_args = self._parser.parse_known_and_unknown_args(args)
        mode = getattr(ns, "assertmode", "plain")
        if mode == "rewrite":
            import _pytest.assertion

            try:
                hook = _pytest.assertion.install_importhook(self)
            except SystemError:
                mode = "plain"
            else:
                self._mark_plugins_for_rewrite(hook)
        self._warn_about_missing_assertion(mode)

    def _mark_plugins_for_rewrite(self, hook) -> None:
        """Given an importhook, mark for rewrite any top-level
        modules or packages in the distribution package for
        all pytest plugins."""
        self.pluginmanager.rewrite_hook = hook

        if os.environ.get("PYTEST_DISABLE_PLUGIN_AUTOLOAD"):
            # We don't autoload from setuptools entry points, no need to continue.
            return

        package_files = (
            str(file)
            for dist in importlib_metadata.distributions()
            if any(ep.group == "pytest11" for ep in dist.entry_points)
            for file in dist.files or []
        )

        for name in _iter_rewritable_modules(package_files):
            hook.mark_rewrite(name)

    def _validate_args(self, args: List[str], via: str) -> List[str]:
        """Validate known args."""
        self._parser._config_source_hint = via  # type: ignore
        try:
            self._parser.parse_known_and_unknown_args(
                args, namespace=copy.copy(self.option)
            )
        finally:
            del self._parser._config_source_hint  # type: ignore

        return args

    def _preparse(self, args: List[str], addopts: bool = True) -> None:
        if addopts:
            env_addopts = os.environ.get("PYTEST_ADDOPTS", "")
            if len(env_addopts):
                args[:] = (
                    self._validate_args(shlex.split(env_addopts), "via PYTEST_ADDOPTS")
                    + args
                )
        self._initini(args)
        if addopts:
            args[:] = (
                self._validate_args(self.getini("addopts"), "via addopts config") + args
            )

        self.known_args_namespace = self._parser.parse_known_args(
            args, namespace=copy.copy(self.option)
        )
        self._checkversion()
        self._consider_importhook(args)
        self.pluginmanager.consider_preparse(args, exclude_only=False)
        if not os.environ.get("PYTEST_DISABLE_PLUGIN_AUTOLOAD"):
            # Don't autoload from setuptools entry point. Only explicitly specified
            # plugins are going to be loaded.
            self.pluginmanager.load_setuptools_entrypoints("pytest11")
        self.pluginmanager.consider_env()

        self.known_args_namespace = self._parser.parse_known_args(
            args, namespace=copy.copy(self.known_args_namespace)
        )

        self._validate_plugins()
        self._warn_about_skipped_plugins()

        if self.known_args_namespace.strict:
            self.issue_config_time_warning(
                _pytest.deprecated.STRICT_OPTION, stacklevel=2
            )

        if self.known_args_namespace.confcutdir is None and self.inipath is not None:
            confcutdir = str(self.inipath.parent)
            self.known_args_namespace.confcutdir = confcutdir
        try:
            self.hook.pytest_load_initial_conftests(
                early_config=self, args=args, parser=self._parser
            )
        except ConftestImportFailure as e:
            if self.known_args_namespace.help or self.known_args_namespace.version:
                # we don't want to prevent --help/--version to work
                # so just let is pass and print a warning at the end
                self.issue_config_time_warning(
                    PytestConfigWarning(f"could not load initial conftests: {e.path}"),
                    stacklevel=2,
                )
            else:
                raise

    @hookimpl(hookwrapper=True)
    def pytest_collection(self) -> Generator[None, None, None]:
        # Validate invalid ini keys after collection is done so we take in account
        # options added by late-loading conftest files.
        yield
        self._validate_config_options()

    def _checkversion(self) -> None:
        import pytest

        minver = self.inicfg.get("minversion", None)
        if minver:
            # Imported lazily to improve start-up time.
            from packaging.version import Version

            if not isinstance(minver, str):
                raise pytest.UsageError(
                    "%s: 'minversion' must be a single value" % self.inipath
                )

            if Version(minver) > Version(pytest.__version__):
                raise pytest.UsageError(
                    "%s: 'minversion' requires pytest-%s, actual pytest-%s'"
                    % (
                        self.inipath,
                        minver,
                        pytest.__version__,
                    )
                )

    def _validate_config_options(self) -> None:
        for key in sorted(self._get_unknown_ini_keys()):
            self._warn_or_fail_if_strict(f"Unknown config option: {key}\n")

    def _validate_plugins(self) -> None:
        required_plugins = sorted(self.getini("required_plugins"))
        if not required_plugins:
            return

        # Imported lazily to improve start-up time.
        from packaging.version import Version
        from packaging.requirements import InvalidRequirement, Requirement

        plugin_info = self.pluginmanager.list_plugin_distinfo()
        plugin_dist_info = {dist.project_name: dist.version for _, dist in plugin_info}

        missing_plugins = []
        for required_plugin in required_plugins:
            try:
                req = Requirement(required_plugin)
            except InvalidRequirement:
                missing_plugins.append(required_plugin)
                continue

            if req.name not in plugin_dist_info:
                missing_plugins.append(required_plugin)
            elif not req.specifier.contains(
                Version(plugin_dist_info[req.name]), prereleases=True
            ):
                missing_plugins.append(required_plugin)

        if missing_plugins:
            raise UsageError(
                "Missing required plugins: {}".format(", ".join(missing_plugins)),
            )

    def _warn_or_fail_if_strict(self, message: str) -> None:
        if self.known_args_namespace.strict_config:
            raise UsageError(message)

        self.issue_config_time_warning(PytestConfigWarning(message), stacklevel=3)

    def _get_unknown_ini_keys(self) -> List[str]:
        parser_inicfg = self._parser._inidict
        return [name for name in self.inicfg if name not in parser_inicfg]

    def parse(self, args: List[str], addopts: bool = True) -> None:
        # Parse given cmdline arguments into this config object.
        assert not hasattr(
            self, "args"
        ), "can only parse cmdline args at most once per Config object"
        self.hook.pytest_addhooks.call_historic(
            kwargs=dict(pluginmanager=self.pluginmanager)
        )
        self._preparse(args, addopts=addopts)
        # XXX deprecated hook:
        self.hook.pytest_cmdline_preparse(config=self, args=args)
        self._parser.after_preparse = True  # type: ignore
        try:
            source = Config.ArgsSource.ARGS
            args = self._parser.parse_setoption(
                args, self.option, namespace=self.option
            )
            if not args:
                if self.invocation_params.dir == self.rootpath:
                    source = Config.ArgsSource.TESTPATHS
                    testpaths: List[str] = self.getini("testpaths")
                    if self.known_args_namespace.pyargs:
                        args = testpaths
                    else:
                        args = []
                        for path in testpaths:
                            args.extend(sorted(glob.iglob(path, recursive=True)))
                if not args:
                    source = Config.ArgsSource.INCOVATION_DIR
                    args = [str(self.invocation_params.dir)]
            self.args = args
            self.args_source = source
        except PrintHelp:
            pass

    def issue_config_time_warning(self, warning: Warning, stacklevel: int) -> None:
        """Issue and handle a warning during the "configure" stage.

        During ``pytest_configure`` we can't capture warnings using the ``catch_warnings_for_item``
        function because it is not possible to have hookwrappers around ``pytest_configure``.

        This function is mainly intended for plugins that need to issue warnings during
        ``pytest_configure`` (or similar stages).

        :param warning: The warning instance.
        :param stacklevel: stacklevel forwarded to warnings.warn.
        """
        if self.pluginmanager.is_blocked("warnings"):
            return

        cmdline_filters = self.known_args_namespace.pythonwarnings or []
        config_filters = self.getini("filterwarnings")

        with warnings.catch_warnings(record=True) as records:
            warnings.simplefilter("always", type(warning))
            apply_warning_filters(config_filters, cmdline_filters)
            warnings.warn(warning, stacklevel=stacklevel)

        if records:
            frame = sys._getframe(stacklevel - 1)
            location = frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name
            self.hook.pytest_warning_recorded.call_historic(
                kwargs=dict(
                    warning_message=records[0],
                    when="config",
                    nodeid="",
                    location=location,
                )
            )

    def addinivalue_line(self, name: str, line: str) -> None:
        """Add a line to an ini-file option. The option must have been
        declared but might not yet be set in which case the line becomes
        the first line in its value."""
        x = self.getini(name)
        assert isinstance(x, list)
        x.append(line)  # modifies the cached list inline

    def getini(self, name: str):
        """Return configuration value from an :ref:`ini file <configfiles>`.

        If the specified name hasn't been registered through a prior
        :func:`parser.addini <pytest.Parser.addini>` call (usually from a
        plugin), a ValueError is raised.
        """
        try:
            return self._inicache[name]
        except KeyError:
            self._inicache[name] = val = self._getini(name)
            return val

    # Meant for easy monkeypatching by legacypath plugin.
    # Can be inlined back (with no cover removed) once legacypath is gone.
    def _getini_unknown_type(self, name: str, type: str, value: Union[str, List[str]]):
        msg = f"unknown configuration type: {type}"
        raise ValueError(msg, value)  # pragma: no cover

    def _getini(self, name: str):
        try:
            description, type, default = self._parser._inidict[name]
        except KeyError as e:
            raise ValueError(f"unknown configuration value: {name!r}") from e
        override_value = self._get_override_ini_value(name)
        if override_value is None:
            try:
                value = self.inicfg[name]
            except KeyError:
                if default is not None:
                    return default
                if type is None:
                    return ""
                return []
        else:
            value = override_value
        # Coerce the values based on types.
        #
        # Note: some coercions are only required if we are reading from .ini files, because
        # the file format doesn't contain type information, but when reading from toml we will
        # get either str or list of str values (see _parse_ini_config_from_pyproject_toml).
        # For example:
        #
        #   ini:
        #     a_line_list = "tests acceptance"
        #   in this case, we need to split the string to obtain a list of strings.
        #
        #   toml:
        #     a_line_list = ["tests", "acceptance"]
        #   in this case, we already have a list ready to use.
        #
        if type == "paths":
            # TODO: This assert is probably not valid in all cases.
            assert self.inipath is not None
            dp = self.inipath.parent
            input_values = shlex.split(value) if isinstance(value, str) else value
            return [dp / x for x in input_values]
        elif type == "args":
            return shlex.split(value) if isinstance(value, str) else value
        elif type == "linelist":
            if isinstance(value, str):
                return [t for t in map(lambda x: x.strip(), value.split("\n")) if t]
            else:
                return value
        elif type == "bool":
            return _strtobool(str(value).strip())
        elif type == "string":
            return value
        elif type is None:
            return value
        else:
            return self._getini_unknown_type(name, type, value)

    def _getconftest_pathlist(
        self, name: str, path: Path, rootpath: Path
    ) -> Optional[List[Path]]:
        try:
            mod, relroots = self.pluginmanager._rget_with_confmod(
                name, path, self.getoption("importmode"), rootpath
            )
        except KeyError:
            return None
        assert mod.__file__ is not None
        modpath = Path(mod.__file__).parent
        values: List[Path] = []
        for relroot in relroots:
            if isinstance(relroot, os.PathLike):
                relroot = Path(relroot)
            else:
                relroot = relroot.replace("/", os.sep)
                relroot = absolutepath(modpath / relroot)
            values.append(relroot)
        return values

    def _get_override_ini_value(self, name: str) -> Optional[str]:
        value = None
        # override_ini is a list of "ini=value" options.
        # Always use the last item if multiple values are set for same ini-name,
        # e.g. -o foo=bar1 -o foo=bar2 will set foo to bar2.
        for ini_config in self._override_ini:
            try:
                key, user_ini_value = ini_config.split("=", 1)
            except ValueError as e:
                raise UsageError(
                    "-o/--override-ini expects option=value style (got: {!r}).".format(
                        ini_config
                    )
                ) from e
            else:
                if key == name:
                    value = user_ini_value
        return value

    def getoption(self, name: str, default=notset, skip: bool = False):
        """Return command line option value.

        :param name: Name of the option.  You may also specify
            the literal ``--OPT`` option instead of the "dest" option name.
        :param default: Default value if no option of that name exists.
        :param skip: If True, raise pytest.skip if option does not exists
            or has a None value.
        """
        name = self._opt2dest.get(name, name)
        try:
            val = getattr(self.option, name)
            if val is None and skip:
                raise AttributeError(name)
            return val
        except AttributeError as e:
            if default is not notset:
                return default
            if skip:
                import pytest

                pytest.skip(f"no {name!r} option found")
            raise ValueError(f"no option named {name!r}") from e

    def getvalue(self, name: str, path=None):
        """Deprecated, use getoption() instead."""
        return self.getoption(name)

    def getvalueorskip(self, name: str, path=None):
        """Deprecated, use getoption(skip=True) instead."""
        return self.getoption(name, skip=True)

    def _warn_about_missing_assertion(self, mode: str) -> None:
        if not _assertion_supported():
            if mode == "plain":
                warning_text = (
                    "ASSERTIONS ARE NOT EXECUTED"
                    " and FAILING TESTS WILL PASS.  Are you"
                    " using python -O?"
                )
            else:
                warning_text = (
                    "assertions not in test modules or"
                    " plugins will be ignored"
                    " because assert statements are not executed "
                    "by the underlying Python interpreter "
                    "(are you using python -O?)\n"
                )
            self.issue_config_time_warning(
                PytestConfigWarning(warning_text),
                stacklevel=3,
            )

    def _warn_about_skipped_plugins(self) -> None:
        for module_name, msg in self.pluginmanager.skipped_plugins:
            self.issue_config_time_warning(
                PytestConfigWarning(f"skipped plugin {module_name!r}: {msg}"),
                stacklevel=2,
            )

def _assertion_supported() -> bool:
    try:
        assert False
    except AssertionError:
        return True
    else:
        return False  # type: ignore[unreachable]

def create_terminal_writer(
    config: Config, file: Optional[TextIO] = None
) -> TerminalWriter:
    """Create a TerminalWriter instance configured according to the options
    in the config object.

    Every code which requires a TerminalWriter object and has access to a
    config object should use this function.
    """
    tw = TerminalWriter(file=file)

    if config.option.color == "yes":
        tw.hasmarkup = True
    elif config.option.color == "no":
        tw.hasmarkup = False

    if config.option.code_highlight == "yes":
        tw.code_highlight = True
    elif config.option.code_highlight == "no":
        tw.code_highlight = False

    return tw

def _strtobool(val: str) -> bool:
    """Convert a string representation of truth to True or False.

    True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
    are 'n', 'no', 'f', 'false', 'off', and '0'.  Raises ValueError if
    'val' is anything else.

    .. note:: Copied from distutils.util.
    """
    val = val.lower()
    if val in ("y", "yes", "t", "true", "on", "1"):
        return True
    elif val in ("n", "no", "f", "false", "off", "0"):
        return False
    else:
        raise ValueError(f"invalid truth value {val!r}")

@lru_cache(maxsize=50)
def parse_warning_filter(
    arg: str, *, escape: bool
) -> Tuple["warnings._ActionKind", str, Type[Warning], str, int]:
    """Parse a warnings filter string.

    This is copied from warnings._setoption with the following changes:

    * Does not apply the filter.
    * Escaping is optional.
    * Raises UsageError so we get nice error messages on failure.
    """
    __tracebackhide__ = True
    error_template = dedent(
        f"""\
        while parsing the following warning configuration:

          {arg}

        This error occurred:

        {{error}}
        """
    )

    parts = arg.split(":")
    if len(parts) > 5:
        doc_url = (
            "https://docs.python.org/3/library/warnings.html#describing-warning-filters"
        )
        error = dedent(
            f"""\
            Too many fields ({len(parts)}), expected at most 5 separated by colons:

              action:message:category:module:line

            For more information please consult: {doc_url}
            """
        )
        raise UsageError(error_template.format(error=error))

    while len(parts) < 5:
        parts.append("")
    action_, message, category_, module, lineno_ = (s.strip() for s in parts)
    try:
        action: "warnings._ActionKind" = warnings._getaction(action_)  # type: ignore[attr-defined]
    except warnings._OptionError as e:
        raise UsageError(error_template.format(error=str(e)))
    try:
        category: Type[Warning] = _resolve_warning_category(category_)
    except Exception:
        exc_info = ExceptionInfo.from_current()
        exception_text = exc_info.getrepr(style="native")
        raise UsageError(error_template.format(error=exception_text))
    if message and escape:
        message = re.escape(message)
    if module and escape:
        module = re.escape(module) + r"\Z"
    if lineno_:
        try:
            lineno = int(lineno_)
            if lineno < 0:
                raise ValueError("number is negative")
        except ValueError as e:
            raise UsageError(
                error_template.format(error=f"invalid lineno {lineno_!r}: {e}")
            )
    else:
        lineno = 0
    return action, message, category, module, lineno

def _resolve_warning_category(category: str) -> Type[Warning]:
    """
    Copied from warnings._getcategory, but changed so it lets exceptions (specially ImportErrors)
    propagate so we can get access to their tracebacks (#9218).
    """
    __tracebackhide__ = True
    if not category:
        return Warning

    if "." not in category:
        import builtins as m

        klass = category
    else:
        module, _, klass = category.rpartition(".")
        m = __import__(module, None, None, [klass])
    cat = getattr(m, klass)
    if not issubclass(cat, Warning):
        raise UsageError(f"{cat} is not a Warning subclass")
    return cast(Type[Warning], cat)

def apply_warning_filters(
    config_filters: Iterable[str], cmdline_filters: Iterable[str]
) -> None:
    """Applies pytest-configured filters to the warnings module"""
    # Filters should have this precedence: cmdline options, config.
    # Filters should be applied in the inverse order of precedence.
    for arg in config_filters:
        warnings.filterwarnings(*parse_warning_filter(arg, escape=False))

    for arg in cmdline_filters:
        warnings.filterwarnings(*parse_warning_filter(arg, escape=True))


text:
{-# LANGUAGE FlexibleInstances #-}

module Unused.ResultsClassifier.Types
    ( LanguageConfiguration(..)
    , LowLikelihoodMatch(..)
    , TermAlias(..)
    , Position(..)
    , Matcher(..)
    , ParseConfigError(..)
    ) where

import qualified Control.Applicative as A
import qualified Control.Monad as M
import qualified Data.HashMap.Strict as HM
import qualified Data.List as L
import           Data.Text (Text)
import qualified Data.Text as T
import           Data.Yaml (FromJSON(..), (.:), (.:?), (.!=))
import qualified Data.Yaml as Y
import           Unused.Projection

data LanguageConfiguration = LanguageConfiguration
    { lcName :: String
    , lcAllowedTerms :: [String]
    , lcAutoLowLikelihood :: [LowLikelihoodMatch]
    , lcTermAliases :: [TermAlias]
    }

data LowLikelihoodMatch = LowLikelihoodMatch
    { smName :: String
    , smMatchers :: [Matcher]
    , smClassOrModule :: Bool
    }

data TermAlias = TermAlias
    { taFrom :: String
    , taTo :: String
    , taTransform :: Text -> Text
    }

data ParseConfigError = ParseConfigError
    { pcePath :: String
    , pceParseError :: String
    }

data Position = StartsWith | EndsWith | Equals
data Matcher = Term Position String | Path Position String | AppOccurrences Int | AllowedTerms [String]

instance FromJSON LanguageConfiguration where
    parseJSON (Y.Object o) = LanguageConfiguration
        <$> o .: "name"
        <*> o .:? "allowedTerms" .!= []
        <*> o .:? "autoLowLikelihood" .!= []
        <*> o .:? "aliases" .!= []
    parseJSON _ = M.mzero

instance FromJSON LowLikelihoodMatch where
    parseJSON (Y.Object o) = LowLikelihoodMatch
        <$> o .: "name"
        <*> parseMatchers o
        <*> o .:? "classOrModule" .!= False
    parseJSON _ = M.mzero

instance FromJSON TermAlias where
    parseJSON (Y.Object o) = TermAlias
        <$> o .: "from"
        <*> o .: "to"
        <*> (either fail return =<< (translate . T.pack <$> (o .: "to")))
    parseJSON _ = M.mzero

data MatchHandler a = MatchHandler
    { mhKeys :: [String]
    , mhKeyToMatcher :: T.Text -> Either T.Text (a -> Matcher)
    }

intHandler :: MatchHandler Int
intHandler = MatchHandler
    { mhKeys = ["appOccurrences"]
    , mhKeyToMatcher = keyToMatcher
    }
  where
    keyToMatcher "appOccurrences" = Right AppOccurrences
    keyToMatcher t                = Left t

stringHandler :: MatchHandler String
stringHandler = MatchHandler
    { mhKeys = ["pathStartsWith", "pathEndsWith", "termStartsWith", "termEndsWith", "termEquals"]
    , mhKeyToMatcher = keyToMatcher
    }
  where
    keyToMatcher "pathStartsWith" = Right $ Path StartsWith
    keyToMatcher "pathEndsWith"   = Right $ Path EndsWith
    keyToMatcher "termStartsWith" = Right $ Term StartsWith
    keyToMatcher "termEndsWith"   = Right $ Term EndsWith
    keyToMatcher "termEquals"     = Right $ Term Equals
    keyToMatcher t                = Left t

stringListHandler :: MatchHandler [String]
stringListHandler = MatchHandler
    { mhKeys = ["allowedTerms"]
    , mhKeyToMatcher = keyToMatcher
    }
  where
    keyToMatcher "allowedTerms" = Right AllowedTerms
    keyToMatcher t              = Left t

lowLikelihoodMatchKeys :: [T.Text]
lowLikelihoodMatchKeys =
    map T.pack $ ["name", "classOrModule"] ++ mhKeys intHandler ++ mhKeys stringHandler ++ mhKeys stringListHandler

validateLowLikelihoodKeys :: Y.Object -> Y.Parser [Matcher] -> Y.Parser [Matcher]
validateLowLikelihoodKeys o ms =
    if fullOverlap
        then ms
        else fail $ "The following keys are unsupported: " ++ L.intercalate ", " (T.unpack <$> unsupportedKeys)
  where
    fullOverlap = null unsupportedKeys
    unsupportedKeys = HM.keys o L.\\ lowLikelihoodMatchKeys

parseMatchers :: Y.Object -> Y.Parser [Matcher]
parseMatchers o =
    validateLowLikelihoodKeys o $ myFold (++) [buildMatcherList o intHandler, buildMatcherList o stringHandler, buildMatcherList o stringListHandler]
  where
    myFold :: (Foldable t, Monad m) => (a -> a -> a) -> t (m a) -> m a
    myFold f = foldl1 (\acc i -> acc >>= (\l -> f l <$> i))

buildMatcherList :: FromJSON a => Y.Object -> MatchHandler a -> Y.Parser [Matcher]
buildMatcherList o mh =
    sequenceA $ matcherParserForKey <$> keysToParse
  where
    matcherParserForKey k = extractMatcher (mhKeyToMatcher mh k) $ mKey k
    keysToParse = positionKeysforMatcher o (mhKeys mh)
    mKey = (.:?) o

positionKeysforMatcher :: Y.Object -> [String] -> [T.Text]
positionKeysforMatcher o ls = L.intersect (T.pack <$> ls) $ HM.keys o

extractMatcher :: Either T.Text (a -> Matcher) -> Y.Parser (Maybe a) -> Y.Parser Matcher
extractMatcher e p = either displayFailure (convertFoundObjectToMatcher p) e

convertFoundObjectToMatcher :: (Monad m, A.Alternative m) => m (Maybe a) -> (a -> b) -> m b
convertFoundObjectToMatcher p f = maybe A.empty (pure . f) =<< p

displayFailure :: T.Text -> Y.Parser a
displayFailure t = fail $ "Parse error: '" ++ T.unpack t ++ "' is not a valid key in a singleOnly matcher"

redpajama_set_name: RedPajamaGithub


text:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Web.Routing;
using NUnit.Framework;

namespace DependencyInjector.Tests
{
    [TestFixture]
    public class RouteConfigTests
    {
        // SetUp & Mocks
        [SetUp]
        public void SetUp()
        {

        }

        [TearDown]
        public void TearDown()
        {

        }

        // Tests
        [Test]
        public void RegisterRoutes_WithNullRouteCollection_ThrowsArgumentNullException()
        {
            Assert.Throws<ArgumentNullException>(() => RouteConfig.RegisterRoutes(null));
        }

        [Test]
        public void RegisterRoutes_WithValidRouteCollection_RegistersIgoreRouteForAxds()
        {
            RouteCollection routes = new RouteCollection();

            RouteConfig.RegisterRoutes(routes);

            Assert.IsNotNull(routes.SingleOrDefault(route=>(route as Route).Url=="{resource}.axd/{*pathInfo}" &&
                (route as Route).RouteHandler is StopRoutingHandler));
        }

        [Test]
        public void RegisterRoutes_WithValidRouteCollection_RegistersDefaultRoute()
        {
            RouteCollection routes = new RouteCollection();

            RouteConfig.RegisterRoutes(routes);

            Route defaultRoute = routes.SingleOrDefault(route=>(route as Route).Url=="{controller}/{action}/{id}") as Route;
            Assert.IsNotNull(defaultRoute);
            Assert.AreEqual("Home", defaultRoute.Defaults["controller"]);
            Assert.AreEqual("Index", defaultRoute.Defaults["action"]);
        }

    }
}


text:
Get all labels for given project.

```
GET /projects/:id/labels
```

```json
[
    {
        "name": "Awesome",
        "color": "#DD10AA"
    },
    {
        "name": "Documentation",
        "color": "#1E80DD"
    },
    {
        "name": "Feature",
        "color": "#11FF22"
    },
    {
        "name": "Bug",
        "color": "#EE1122"
    }
]
```

## Create a new label

Creates a new label for given repository with given name and color.

```
POST /projects/:id/labels
```

Parameters:

- `id` (required) - The ID of a project
- `name` (required) - The name of the label
- `color` (required) -  Color of the label given in 6-digit hex notation with leading '#' sign (e.g. #FFAABB)

It returns 200 and the newly created label, if the operation succeeds.
If the label already exists, 409 and an error message is returned.
If label parameters are invalid, 400 and an explaining error message is returned.

## Delete a label

Deletes a label given by its name.

```
DELETE /projects/:id/labels
```

- `id` (required) - The ID of a project
- `name` (required) - The name of the label to be deleted

It returns 200 if the label successfully was deleted, 400 for wrong parameters
and 404 if the label does not exist.
In case of an error, additionally an error message is returned.

## Edit an existing label

Updates an existing label with new name or now color. At least one parameter
is required, to update the label.

```
PUT /projects/:id/labels
```

Parameters:

- `id` (required) - The ID of a project
- `name` (required) - The name of the existing label
- `new_name` (optional) - The new name of the label
- `color` (optional) -  New color of the label given in 6-digit hex notation with leading '#' sign (e.g. #FFAABB)

On success, this method returns 200 with the updated label.
If required parameters are missing or parameters are invalid, 400 is returned.
If the label to be updated is missing, 404 is returned.
In case of an error, additionally an error message is returned.


text:
'''All settings are instantiated here:
    ISO currency codes dict
    Exchange Rate sources
'''

import urllib2 as http
import xml.etree.ElementTree as xml
from pkg_resources import resource_string

class CurrencyCode():
    '''Class contains methods to get currency codes from ISO 4217 list
    It uses package data file 'data/iso_4217.xml' if filename not provided
    If package data file is missing or empty then it tries to download list
    from website and save it to filename if provided
    '''

    def __init__(self, filename=None):
        '''build dict of ISO currency codes:
            numeric codes are used in database
            literal codes are used in WS calls and responses

        Keyword arguments:
            filename -- path to xml list with currency codes
        '''
        self._ccy_codes = {}

        try:
            s = resource_string(__name__, 'data/iso_4217.xml')
        except IOError as err:
            # try to load from website
            url = 'http://www.currency-iso.org/dam/downloads/lists/list_one.xml'
            s = http.urlopen(url).read()

            # try to save to file if provided
            if filename is not None:
                with open(filename, 'w') as f:
                    f.write(s)

        if s:
            # build dict using comprehension
            self._ccy_codes = {
                ccy_entry[2].text: {
                    'cntry_name': ccy_entry[0].text,
                    'ccy_name': ccy_entry[1].text,
                    'ccy_code': int(ccy_entry[3].text),
                    'ccy_units': ccy_entry[4].text
                }
                for ccy_entry
                in xml.fromstring(s).iter('CcyNtry')
                if len(ccy_entry) == 5
            }

    def get_ccy_num_code(self, ccy_char_code):
        '''returns currency ISO 4217 numeric code by character code'''
        return self._ccy_codes.get(ccy_char_code, {}).get('ccy_code', -1)

    def get_ccy_num_codes(self, ccy_char_codes):
        '''returns dict with char_code : num_code values'''
        return {ccy_char_code: self._ccy_codes[ccy_char_code]['ccy_code']
                for ccy_char_code
                in ccy_char_codes
                if ccy_char_code in self._ccy_codes.keys()
               }

    def get_ccy_codes(self):
        '''returns literal: digit code pairs'''
        return {k: v['ccy_code'] for k, v in self._ccy_codes.iteritems()}

    def get_ccy_info(self, ccy_char_code):
        '''returns all currency info from ISO 4217 table by character code'''
        return self._ccy_codes.get(ccy_char_code, {})

class ExchangeRateSource():
    '''dictionary containing information about supported sources.
    Config fields:
        key -- string key used for lookup in python
        id -- integer identifier. used in database table
        url -- string representation of source url (contains formatting points)
        dateformat -- format of date used for constructing source URL
        datapoints -- set of mandatory field names in JSON response (for
            informational purpose only)
        max_connections -- number of concurrent requests to be made to WS
        field_mapper -- method name for generating output from JSON
    '''

    def __init__(self):
        '''initiate exrate source config dict
        rewrite this method if config needs to be moved to file or database'''

        self._exrate_sources = {
                'NBU-json':
                {
                            'id': 1,
                            'url': 'https://bank.gov.ua/NBUStatService/v1/' +
                                'statdirectory/exchange?' +
                                'date={exdate}&valcode={basecur}&json',
                            'dateformat': '%Y%m%d',
                            'datapoints': {'r030', 'rate', 'exchangedate'},
                            'max_connections': 10,
                            'field_mapper': '_map_nbu_gov_ua'
                        },
                'ECB-Fixer':
                {
                            'id': 2,
                            'url': 'https://api.fixer.io/' +
                                '{exdate}?base={basecur}&symbols={localcur}',
                            'dateformat': '%Y-%m-%d',
                            'datapoints': {'base', 'date', 'rates'},
                            'max_connections': 10,
                            'field_mapper': '_map_ecb_fixer'
                        }
        }

    def get_source_config(self, exratesrc):
        '''returns dict containing information for provided source'''

        return self._exrate_sources.get(exratesrc, {})


text:
"""
Update reference results for static analyzer.
"""

import SATestBuild

from subprocess import check_call
import os
import sys

Verbose = 1

def runCmd(Command, **kwargs):
    if Verbose:
        print "Executing %s" % Command
    check_call(Command, shell=True, **kwargs)

def updateReferenceResults(ProjName, ProjBuildMode):
    ProjDir = SATestBuild.getProjectDir(ProjName)

    RefResultsPath = os.path.join(
        ProjDir,
        SATestBuild.getSBOutputDirName(IsReferenceBuild=True))
    CreatedResultsPath = os.path.join(
        ProjDir,
        SATestBuild.getSBOutputDirName(IsReferenceBuild=False))

    if not os.path.exists(CreatedResultsPath):
        print >> sys.stderr, "New results not found, was SATestBuild.py "\
                             "previously run?"
        sys.exit(1)

    BuildLogPath = SATestBuild.getBuildLogPath(RefResultsPath)
    Dirname = os.path.dirname(os.path.abspath(BuildLogPath))
    runCmd("mkdir -p '%s'" % Dirname)
    with open(BuildLogPath, "wb+") as PBuildLogFile:
        # Remove reference results: in git, and then again for a good measure
        # with rm, as git might not remove things fully if there are empty
        # directories involved.
        runCmd('git rm -r -q "%s"' % (RefResultsPath,), stdout=PBuildLogFile)
        runCmd('rm -rf "%s"' % (RefResultsPath,), stdout=PBuildLogFile)

        # Replace reference results with a freshly computed once.
        runCmd('cp -r "%s" "%s"' % (CreatedResultsPath, RefResultsPath,),
               stdout=PBuildLogFile)

        # Run cleanup script.
        SATestBuild.runCleanupScript(ProjDir, PBuildLogFile)

        SATestBuild.normalizeReferenceResults(
            ProjDir, RefResultsPath, ProjBuildMode)

        # Clean up the generated difference results.
        SATestBuild.cleanupReferenceResults(RefResultsPath)

        runCmd('git add "%s"' % (RefResultsPath,), stdout=PBuildLogFile)

def main(argv):
    if len(argv) == 2 and argv[1] in ('-h', '--help'):
        print >> sys.stderr, "Update static analyzer reference results based "\
                             "\non the previous run of SATestBuild.py.\n"\
                             "\nN.B.: Assumes that SATestBuild.py was just run"
        sys.exit(1)

    with SATestBuild.projectFileHandler() as f:
        for (ProjName, ProjBuildMode) in SATestBuild.iterateOverProjects(f):
            updateReferenceResults(ProjName, int(ProjBuildMode))

if __name__ == '__main__':
    main(sys.argv)


text:
from iris_sdk.models.base_resource import BaseResource
from iris_sdk.models.data.site_totaltns import SiteTotaltnsData

XML_NAME_TOTALTNS_SITE = "SiteTNs"
XPATH_TOTALTNS_SITE = "/totaltns"

class SiteTotaltns(BaseResource, SiteTotaltnsData):

    """Total telephone numbers count for a site"""

    _node_name = XML_NAME_TOTALTNS_SITE
    _xpath = XPATH_TOTALTNS_SITE

    def get(self):
        return self._get_data()


text:
package com.sage42.android.view.fonts;

import android.content.Context;
import android.graphics.Typeface;
import android.inputmethodservice.ExtractEditText;
import android.util.AttributeSet;

public class MyExtractEditText extends ExtractEditText
{

    public MyExtractEditText(final Context context, final AttributeSet attrs, final int defStyle)
    {
        super(context, attrs, defStyle);
        this.initCustomFonts(context, attrs);
    }

    public MyExtractEditText(final Context context, final AttributeSet attrs)
    {
        super(context, attrs);
        this.initCustomFonts(context, attrs);
    }

    public MyExtractEditText(final Context context)
    {
        super(context);
        this.initCustomFonts(context, null);
    }

    /**
     * Extract any custom font related settings from supplied args.
     *
     * @param context
     * @param attrs
     */
    private void initCustomFonts(final Context context, final AttributeSet attrs)
    {
        if (this.isInEditMode())
        {
            // this is preview mode so we need to stop processing
            return;
        }

        // Fonts work as a combination of particular family and the style.
        final Typeface typeface = FontManager.extractTypeface(context, attrs);
        if (typeface != null)
        {
            this.setTypeface(typeface);
        }
    }

}


text:
<?php

namespace spec\DigitalKaoz\TTD\Parser;

use DigitalKaoz\TTD\Parser\NodeFilter;
use PhpParser\Node\Param;
use PhpParser\Node\Stmt\ClassMethod;
use PhpParser\Node\Stmt\Namespace_;
use PhpSpec\ObjectBehavior;

class NodeFilterSpec extends ObjectBehavior
{
    public function it_is_initializable()
    {
        $this->shouldHaveType(NodeFilter::class);
    }

    public function it_can_detect_namespace_nodes()
    {
        $this->isNamespace(new Namespace_())->shouldBe(true);
        $this->isNamespace(new ClassMethod('foo'))->shouldBe(false);
    }

    public function it_can_detect_class_methods()
    {
        $this->isMethod(new Namespace_())->shouldBe(false);
        $this->isMethod(new ClassMethod('foo'))->shouldBe(true);
    }

    public function it_can_detect_methods_with_typed_parameters()
    {
        $this->hasTypedParameters(new Namespace_())->shouldBe(false);
        $this->hasTypedParameters(new ClassMethod('foo'))->shouldBe(false);

        $this->hasTypedParameters(new ClassMethod('foo', ['params' => [new Param('foo', null, 'Foo')]]))->shouldBe(true);
    }

    public function it_can_match_methods_by_name()
    {
        $this->beConstructedWith('/^[let|go|it_].*$/');

        $this->matchesFilterPattern(new Namespace_())->shouldBe(false);
        $this->matchesFilterPattern(new  ClassMethod('foo'))->shouldBe(false);

        $this->matchesFilterPattern(new ClassMethod('let'))->shouldBe(true);
        $this->matchesFilterPattern(new ClassMethod('go'))->shouldBe(true);
        $this->matchesFilterPattern(new ClassMethod('it_solves'))->shouldBe(true);
    }
}


text:
package org.archive.wayback.accesscontrol.robotstxt.redis;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.io.IOUtils;
import org.archive.wayback.accesscontrol.robotstxt.redis.RedisRobotsLogic.RedisValue;
import org.archive.wayback.core.Resource;
import org.archive.wayback.exception.LiveDocumentNotAvailableException;
import org.archive.wayback.exception.LiveWebCacheUnavailableException;
import org.archive.wayback.exception.LiveWebTimeoutException;
import org.archive.wayback.liveweb.LiveWebCache;
import org.archive.wayback.webapp.PerfStats;

import com.google.common.io.ByteStreams;

public class SimpleRedisRobotsCache implements LiveWebCache {

	private final static Logger LOGGER = Logger
	.getLogger(SimpleRedisRobotsCache.class.getName());

	enum PerfStat
	{
		RobotsRedis,
		RobotsLive,
	};

	/* REDIS */
	protected RedisRobotsLogic redisCmds;

	/* EXTERNAL CACHE */
	protected LiveWebCache liveweb = null;

	protected boolean gzipRobots = false;

	final static int STATUS_OK = 200;
	final static int STATUS_ERROR = 502;

	final static int MAX_ROBOTS_SIZE = 500000;

	final static int ONE_DAY = 60 * 60 * 24;

	private int totalTTL = ONE_DAY * 10;
	private int refreshTTL = ONE_DAY;

	private int notAvailTotalTTL = ONE_DAY * 2;
	private int notAvailRefreshTTL = ONE_DAY / 2;

	final static String ROBOTS_TOKEN_EMPTY = "0_ROBOTS_EMPTY";

	final static String ROBOTS_TOKEN_ERROR = "0_ROBOTS_ERROR-";
	final static String ROBOTS_TOKEN_ERROR_UNKNOWN = "0_ROBOTS_ERROR-0";

	final static String UPDATE_QUEUE_KEY = "robots_update_queue";
	final static int MAX_UPDATE_QUEUE_SIZE = 50000;

	@Override
	public Resource getCachedResource(URL urlURL, long maxCacheMS,
				boolean bUseOlder) throws LiveDocumentNotAvailableException,
				LiveWebCacheUnavailableException, LiveWebTimeoutException,
				IOException {

		String url = urlURL.toExternalForm();

		RedisValue value = null;

		try {
			PerfStats.timeStart(PerfStat.RobotsRedis);

			if (redisCmds != null) {
				value = redisCmds.getValue(url);
			}
		} catch (LiveWebCacheUnavailableException lw) {
			value = null;
		} finally {
			PerfStats.timeEnd(PerfStat.RobotsRedis);
		}

		// Use the old liveweb cache, if provided
		if (value == null) {
			RobotsResult result = loadExternal(urlURL, maxCacheMS, bUseOlder);

			PerfStats.timeStart(PerfStat.RobotsRedis);
			this.updateCache(result.robots, url, null, result.status, true);
			PerfStats.timeEnd(PerfStat.RobotsRedis);

			if (result == null || result.status != STATUS_OK) {
				throw new LiveDocumentNotAvailableException("Error Loading Live Robots");
			}

			return new RobotsTxtResource(result.robots);

		} else {

			if (isExpired(value, url, 0)) {
				PerfStats.timeStart(PerfStat.RobotsRedis);
				redisCmds.pushKey(UPDATE_QUEUE_KEY, url, MAX_UPDATE_QUEUE_SIZE);
				PerfStats.timeEnd(PerfStat.RobotsRedis);
			}

			String currentRobots = value.value;

			if (currentRobots.startsWith(ROBOTS_TOKEN_ERROR)) {
				throw new LiveDocumentNotAvailableException("Robots Error: " + currentRobots);
			} else if (value.equals(ROBOTS_TOKEN_EMPTY)) {
				currentRobots = "";
			}

			return new RobotsTxtResource(currentRobots);
		}
	}

	@Override
	public void shutdown() {
		if (redisCmds != null) {
			redisCmds.close();
			redisCmds = null;
		}
	}

	static boolean isFailedError(int status)
	{
		return (status == 0) || ((status >= 500));
	}

	static boolean isRedirect(int status)
	{
		return (status == 301) || (status == 302);
	}

	static boolean isFailedError(String code)
	{
		try {
			int status = Integer.parseInt(code);
			return isFailedError(status);
		} catch (NumberFormatException n) {
			return true;
		}
	}

	public boolean isExpired(RedisValue value, String url, int customRefreshTime) {

		int maxTime, refreshTime;

		boolean isFailedError = value.value.startsWith(ROBOTS_TOKEN_ERROR);

		if (isFailedError) {
			String code = value.value.substring(ROBOTS_TOKEN_ERROR.length());
			isFailedError = isFailedError(code);
		}

		if (isFailedError) {
			maxTime = notAvailTotalTTL;
			refreshTime = notAvailRefreshTTL;
		} else {
			maxTime = totalTTL;
			refreshTime = refreshTTL;
		}

		if (customRefreshTime > 0) {
			refreshTime = customRefreshTime;
		}

		if ((maxTime - value.ttl) >= refreshTime) {
//			LOGGER.info("Queue for robot refresh: "
//					+ (maxTime - value.ttl) + ">=" + refreshTime + " " + url);

			return true;
		}

		return false;
	}

	class RobotsResult
	{
		String oldRobots;
		String robots;
		int status;

		RobotsResult(String oldRobots)
		{
			status = 0;
			this.oldRobots = oldRobots;
			robots = null;
		}

		RobotsResult(String robots, int status)
		{
			this.robots = robots;
			this.status = status;
		}

		boolean isSameRobots()
		{
			return (robots == null) || (oldRobots == null) || robots.equals(oldRobots);
		}
	}

	protected RobotsResult loadExternal(URL urlURL, long maxCacheMS, boolean bUseOlder)
	{
		//RobotsContext context = new RobotsContext(url, current, true, true);

		Resource origResource = null;
		int status = 0;
		String contents = null;

		try {
			PerfStats.timeStart(PerfStat.RobotsLive);

			origResource = liveweb.getCachedResource(urlURL, maxCacheMS, bUseOlder);

			status = origResource.getStatusCode();

			if (status == STATUS_OK) {
				if (origResource instanceof RobotsTxtResource) {
					contents = ((RobotsTxtResource)origResource).getContents();
				} else {
					contents = IOUtils.toString(ByteStreams.limit(origResource, MAX_ROBOTS_SIZE), "UTF-8");
				}
			}
		} catch (Exception e) {
			status = STATUS_ERROR;
		} finally {
			if (origResource != null) {
				try {
					origResource.close();
				} catch (IOException e) {

				}
			}
			PerfStats.timeEnd(PerfStat.RobotsLive);
		}

		return new RobotsResult(contents, status);
	}

	protected void updateCache(final String contents, final String url, final String current, int status, boolean cacheFails) {

		String newRedisValue = null;
		int newTTL = 0;
		boolean ttlOnly = false;

		if (status == STATUS_OK) {
			//contents = context.getNewRobots();
			newTTL = totalTTL;

			if (contents == null || contents.isEmpty()) {
				newRedisValue = ROBOTS_TOKEN_EMPTY;
			} else if (contents.length() > MAX_ROBOTS_SIZE) {
				newRedisValue = contents.substring(0, MAX_ROBOTS_SIZE);
			} else {
				newRedisValue = contents;
			}

		} else {
			if (isFailedError(status)) {
				newTTL = notAvailTotalTTL;

				// Only Cacheing successful lookups
				if (!cacheFails) {
					return;
				}
			} else {
				newTTL = totalTTL;
			}

			newRedisValue = ROBOTS_TOKEN_ERROR + status;
		}

		String currentValue = current;

		if (currentValue != null) {
			if (currentValue.equals(newRedisValue)) {
				ttlOnly = true;
			}

			// Don't override a valid robots with a timeout error
			if (!isRedirect(status) && !isValidRobots(newRedisValue) && isValidRobots(currentValue)) {
				newTTL = totalTTL;
				ttlOnly = true;
				if (LOGGER.isLoggable(Level.INFO)) {
					LOGGER.info("REFRESH ERROR: " + status + " - Keeping same robots for " + url);
				}
			}
		}

		final RedisValue value = new RedisValue((ttlOnly ? null : newRedisValue), newTTL);

		redisCmds.updateValue(url, value, gzipRobots);
	}

	protected boolean isValidRobots(String value) {
		return !value.startsWith(ROBOTS_TOKEN_ERROR) && !value.equals(ROBOTS_TOKEN_EMPTY);
	}

	public RobotsResult forceUpdate(String url, int minUpdateTime, boolean cacheFails)
	{
		String current = null;

		try {
			RedisValue value = null;

			try {
				PerfStats.timeStart(PerfStat.RobotsRedis);
				value = redisCmds.getValue(url);
			} finally {
				PerfStats.timeEnd(PerfStat.RobotsRedis);
			}

			// Just in case, avoid too many updates
			if ((minUpdateTime > 0) && (value != null) && !isExpired(value, url, minUpdateTime)) {
				return new RobotsResult(value.value);
			}

			current = (value != null ? value.value : null);
		} catch (LiveWebCacheUnavailableException lw) {
			current = lw.toString();
		}

		RobotsResult result = null;

		try {
			result = loadExternal(new URL(url), 0, false);
		} catch (MalformedURLException e) {
			return new RobotsResult(current);
		}

		if ((result.status == STATUS_OK) || cacheFails) {
			this.updateCache(result.robots, url, current, result.status, cacheFails);

//			if (LOGGER.isLoggable(Level.INFO)) {
//				LOGGER.info("Force updated: " + url);
//			}
		}

		result.oldRobots = current;
		return result;
	}

//	public RedisRobotsLogic getRedisConnMan() {
//		return redisCmds.;
//	}

	public void setRedisConnMan(RedisConnectionManager redisConn) {
		this.redisCmds = new RedisRobotsLogic(redisConn);
	}

	public LiveWebCache getLiveweb() {
		return liveweb;
	}

	public void setLiveweb(LiveWebCache liveweb) {
		this.liveweb = liveweb;
	}

	public boolean isGzipRobots() {
		return gzipRobots;
	}

	public void setGzipRobots(boolean gzipRobots) {
		this.gzipRobots = gzipRobots;
	}
}


text:
from devices.models import Configuration, Platform
from utils.testing import ViewTestCases

class ConfigurationTestCase(ViewTestCases.PrimaryObjectViewTestCase):
    model = Configuration

    test_bulk_edit_objects = None

    @classmethod
    def setUpTestData(cls):
        Configuration.objects.bulk_create(
            [
                Configuration(name="Configuration 1", template="Configuration 1"),
                Configuration(name="Configuration 2", template="Configuration 2"),
                Configuration(name="Configuration 3", template="Configuration 3"),
            ]
        )

        cls.form_data = {
            "name": "Configuration 4",
            "template": "Configuration 4",
            "comments": "",
            "tags": [],
        }

class PlatformTestCase(
    ViewTestCases.CreateObjectViewTestCase,
    ViewTestCases.EditObjectViewTestCase,
    ViewTestCases.DeleteObjectViewTestCase,
    ViewTestCases.ListObjectsViewTestCase,
):
    model = Platform

    @classmethod
    def setUpTestData(cls):
        Platform.objects.bulk_create(
            [
                Platform(name="Some OS", slug="someos"),
                Platform(name="Test OS", slug="testos"),
            ]
        )

        cls.form_data = {
            "name": "Bugs OS",
            "slug": "bugsos",
            "napalm_driver": "bugsos",
            "napalm_args": {},
            "password_algorithm": "",
            "description": "",
        }


text:
module.exports = {
  db: {
    username: process.env.EGAP_DB_USER || 'node',
    password: process.env.EGAP_DB_PASSWORD || 'TestDatabasePassword',
    address: process.env.EGAP_DB_ADDRESS || 'localhost',
    database: 'electrongap'
  },
  api: {
    port: 3000
  },
  web: {
    port: 4000
  }
};


text:
package org.jbpm.workbench.client;

import com.google.gwt.core.client.GWT;
import com.google.gwt.resources.client.ClientBundle;
import org.jbpm.workbench.client.images.AppImages;

public interface AppResource
        extends
        ClientBundle {

    AppResource INSTANCE = GWT.create(AppResource.class);

    AppImages images();
}