# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
# For details: https://github.com/coveragepy/coveragepy/blob/main/NOTICE.txt

"""Control of and utilities for debugging."""

from __future__ import annotations

import _thread
import atexit
import contextlib
import datetime
import functools
import inspect
import itertools
import os
import pprint
import re
import reprlib
import sys
import traceback
import types
from collections.abc import Iterable, Iterator, Mapping
from typing import IO, Any, Callable, Final, overload

from coverage.misc import human_sorted_items, isolate_module
from coverage.types import AnyCallable, TWritable

os = isolate_module(os)


# When debugging, it can be helpful to force some options, especially when
# debugging the configuration mechanisms you usually use to control debugging!
# This is a list of forced debugging options.
FORCED_DEBUG: list[str] = []
FORCED_DEBUG_FILE = None


class DebugControl:
    """Control and output for debugging."""

    show_repr_attr = False  # For auto_repr

    def __init__(
        self,
        options: Iterable[str],
        output: IO[str] | None,
        file_name: str | None = None,
    ) -> None:
        """Configure the options and output file for debugging."""
        self.options = list(options) + FORCED_DEBUG
        self.suppress_callers = False

        filters = []
        if self.should("process"):
            filters.append(CwdTracker().filter)
            filters.append(ProcessTracker().filter)
        if self.should("pytest"):
            filters.append(PytestTracker().filter)
        if self.should("pid"):
            filters.append(add_pid_and_tid)

        self.output = DebugOutputFile.get_one(
            output,
            file_name=file_name,
            filters=filters,
        )
        self.raw_output = self.output.outfile

    def __repr__(self) -> str:
        return f"<DebugControl options={self.options!r} raw_output={self.raw_output!r}>"

    def should(self, option: str) -> bool:
        """Decide whether to output debug information in category `option`."""
        if option == "callers" and self.suppress_callers:
            return False
        return option in self.options

    @contextlib.contextmanager
    def without_callers(self) -> Iterator[None]:
        """A context manager to prevent call stacks from being logged."""
        old = self.suppress_callers
        self.suppress_callers = True
        try:
            yield
        finally:
            self.suppress_callers = old

    def write(self, msg: str, *, exc: BaseException | None = None) -> None:
        """Write a line of debug output.

        `msg` is the line to write. A newline will be appended.

        If `exc` is provided, a stack trace of the exception will be written
        after the message.

        """
        self.output.write(msg + "\n")
        if exc is not None:
            self.output.write("".join(traceback.format_exception(None, exc, exc.__traceback__)))
        if self.should("self"):
            caller_self = inspect.stack()[1][0].f_locals.get("self")
            if caller_self is not None:
                self.output.write(f"self: {caller_self!r}\n")
        if self.should("callers"):
            dump_stack_frames(out=self.output, skip=1)
        self.output.flush()


class NoDebugging(DebugControl):
    """A replacement for DebugControl that will never try to do anything."""

    def __init__(self) -> None:
        # pylint: disable=super-init-not-called
        pass

    def should(self, option: str) -> bool:
        """Should we write debug messages?  Never."""
        return False

    @contextlib.contextmanager
    def without_callers(self) -> Iterator[None]:
        """A dummy context manager to satisfy the api."""
        yield  # pragma: never called

    def write(self, msg: str, *, exc: BaseException | None = None) -> None:
        """This will never be called."""
        raise AssertionError("NoDebugging.write should never be called.")


class DevNullDebug(NoDebugging):
    """A DebugControl that won't write anywhere."""

    def write(self, msg: str, *, exc: BaseException | None = None) -> None:
        pass


def info_header(label: str) -> str:
    """Make a nice header string."""
    return "--{:-<60s}".format(" " + label + " ")


def info_formatter(info: Iterable[tuple[str, Any]]) -> Iterable[str]:
    """Produce a sequence of formatted lines from info.

    `info` is a sequence of pairs (label, data).  The produced lines are
    nicely formatted, ready to print.

    """
    info = list(info)
    if not info:
        return
    LABEL_LEN = 30
    assert all(len(l) < LABEL_LEN for l, _ in info)
    for label, data in info:
        if data == []:
            data = "-none-"
        prefix = f"{label:>{LABEL_LEN}}: "
        match data:
            case tuple() if len(str(data)) < 30:
                yield f"{prefix}{data}"
            case tuple() | list() | set():
                for e in data:
                    yield f"{prefix}{e}"
                    prefix = " " * (LABEL_LEN + 2)
            case _:
                yield f"{prefix}{data}"


def write_formatted_info(
    write: Callable[[str], None],
    header: str,
    info: Iterable[tuple[str, Any]],
) -> None:
    """Write a sequence of (label,data) pairs nicely.

    `write` is a function write(str) that accepts each line of output.
    `header` is a string to start the section.  `info` is a sequence of
    (label, data) pairs, where label is a str, and data can be a single
    value, or a list/set/tuple.

    """
    write(info_header(header))
    for line in info_formatter(info):
        write(f" {line}")


def exc_one_line(exc: Exception) -> str:
    """Get a one-line summary of an exception, including class name and message."""
    lines = traceback.format_exception_only(type(exc), exc)
    return "|".join(l.rstrip() for l in lines)


_FILENAME_REGEXES: list[tuple[str, str]] = [
    (r".*[/\\]pytest-of-.*[/\\]pytest-\d+([/\\]popen-gw\d+)?", "tmp:"),
]
_FILENAME_SUBS: list[tuple[str, str]] = []


@overload
def short_filename(filename: str) -> str:
    pass


@overload
def short_filename(filename: None) -> None:
    pass


def short_filename(filename: str | None) -> str | None:
    """Shorten a file name. Directories are replaced by prefixes like 'syspath:'"""
    if not _FILENAME_SUBS:
        for pathdir in sys.path:
            _FILENAME_SUBS.append((pathdir, "syspath:"))
        import coverage

        _FILENAME_SUBS.append((os.path.dirname(coverage.__file__), "cov:"))
        _FILENAME_SUBS.sort(key=(lambda pair: len(pair[0])), reverse=True)
    if filename is not None:
        for pat, sub in _FILENAME_REGEXES:
            filename = re.sub(pat, sub, filename)
        for before, after in _FILENAME_SUBS:
            filename = filename.replace(before, after)
    return filename


def file_summary(filename: str) -> str:
    """A one-line summary of a file, for log messages."""
    try:
        s = os.stat(filename)
    except FileNotFoundError:
        summary = "does not exist"
    except Exception as e:
        summary = f"error: {e}"
    else:
        mod = datetime.datetime.fromtimestamp(s.st_mtime)
        summary = f"{s.st_size} bytes, modified {mod}"
    return summary


def short_stack(
    skip: int = 0,
    full: bool = False,
    frame_ids: bool = False,
    short_filenames: bool = False,
) -> str:
    """Return a string summarizing the call stack.

    The string is multi-line, with one line per stack frame. Each line shows
    the function name, the file name, and the line number:

        ...
        start_import_stop : /Users/ned/coverage/trunk/tests/coveragetest.py:95
        import_local_file : /Users/ned/coverage/trunk/tests/coveragetest.py:81
        import_local_file : /Users/ned/coverage/trunk/coverage/backward.py:159
        ...

    `skip` is the number of closest immediate frames to skip, so that debugging
    functions can call this and not be included in the result.

    If `full` is true, then include all frames.  Otherwise, initial "boring"
    frames (ones in site-packages and earlier) are omitted.

    `short_filenames` will shorten filenames using `short_filename`, to reduce
    the amount of repetitive noise in stack traces.

    """
    # Regexes in initial frames that we don't care about.
    # fmt: off
    BORING_PRELUDE = [
        "<string>",             # pytest-xdist has string execution.
        r"\bigor.py$",          # Our test runner.
        r"\bsite-packages\b",   # pytest etc getting to our tests.
    ]
    # fmt: on

    stack: Iterable[inspect.FrameInfo] = inspect.stack()[:skip:-1]
    if not full:
        for pat in BORING_PRELUDE:
            stack = itertools.dropwhile(
                (lambda fi, pat=pat: re.search(pat, fi.filename)),  # type: ignore[misc]
                stack,
            )
    lines = []
    for frame_info in stack:
        line = f"{frame_info.function:>30s} : "
        if frame_ids:
            line += f"{id(frame_info.frame):#x} "
        filename = frame_info.filename
        if short_filenames:
            filename = short_filename(filename)
        line += f"{filename}:{frame_info.lineno}"
        lines.append(line)
    return "\n".join(lines)


def dump_stack_frames(out: TWritable, skip: int = 0) -> None:
    """Print a summary of the stack to `out`."""
    out.write(short_stack(skip=skip + 1) + "\n")


def clipped_repr(text: str, numchars: int = 50) -> str:
    """`repr(text)`, but limited to `numchars`."""
    r = reprlib.Repr()
    r.maxstring = numchars
    return r.repr(text)


def short_id(id64: int) -> int:
    """Given a 64-bit id, make a shorter 16-bit one."""
    id16 = 0
    for offset in range(0, 64, 16):
        id16 ^= id64 >> offset
    return id16 & 0xFFFF


def add_pid_and_tid(text: str) -> str:
    """A filter to add pid and tid to debug messages."""
    # Thread ids are useful, but too long. Make a shorter one.
    tid = f"{short_id(_thread.get_ident()):04x}"
    text = f"{os.getpid():5d}.{tid}: {text}"
    return text


AUTO_REPR_IGNORE = {"$coverage.object_id"}


def auto_repr(self: Any) -> str:
    """A function implementing an automatic __repr__ for debugging."""
    show_attrs = (
        (k, v)
        for k, v in self.__dict__.items()
        if getattr(v, "show_repr_attr", True)
        and not inspect.ismethod(v)
        and k not in AUTO_REPR_IGNORE
    )
    return "<{klass} @{id:#x}{attrs}>".format(
        klass=self.__class__.__name__,
        id=id(self),
        attrs="".join(f" {k}={v!r}" for k, v in show_attrs),
    )


def simplify(v: Any) -> Any:  # pragma: debugging
    """Turn things which are nearly dict/list/etc into dict/list/etc."""
    if isinstance(v, dict):
        return {k: simplify(vv) for k, vv in v.items()}
    elif isinstance(v, (list, tuple)):
        return type(v)(simplify(vv) for vv in v)
    elif hasattr(v, "__dict__"):
        return simplify({"." + k: v for k, v in v.__dict__.items()})
    else:
        return v


def ppformat(v: Any) -> str:  # pragma: debugging
    """Debug helper to pretty-print data, including SimpleNamespace objects."""
    return pprint.pformat(simplify(v), indent=4, compact=True, sort_dicts=True, width=140)


def pp(v: Any) -> None:  # pragma: debugging
    """Debug helper to pretty-print data, including SimpleNamespace objects."""
    print(ppformat(v))


def filter_text(text: str, filters: Iterable[Callable[[str], str]]) -> str:
    """Run `text` through a series of filters.

    `filters` is a list of functions. Each takes a string and returns a
    string.  Each is run in turn. After each filter, the text is split into
    lines, and each line is passed through the next filter.

    Returns: the final string that results after all of the filters have
    run.

    """
    clean_text = text.rstrip()
    ending = text[len(clean_text) :]
    text = clean_text
    for filter_fn in filters:
        lines = []
        for line in text.splitlines():
            lines.extend(filter_fn(line).splitlines())
        text = "\n".join(lines)
    return text + ending


class CwdTracker:
    """A class to add cwd info to debug messages."""

    def __init__(self) -> None:
        self.cwd: str | None = None

    def filter(self, text: str) -> str:
        """Add a cwd message for each new cwd."""
        cwd = os.getcwd()
        if cwd != self.cwd:
            text = f"cwd is now {cwd!r}\n{text}"
            self.cwd = cwd
        return text


class ProcessTracker:
    """Track process creation for debug logging."""

    def __init__(self) -> None:
        self.pid: int = os.getpid()
        self.did_welcome = False

    def filter(self, text: str) -> str:
        """Add a message about how new processes came to be."""
        welcome = ""
        pid = os.getpid()
        if self.pid != pid:
            welcome = f"New process: forked {self.pid} -> {pid}\n"
            self.pid = pid
        elif not self.did_welcome:
            argv = getattr(sys, "argv", None)
            welcome = (
                f"New process: {pid=}, executable: {sys.executable!r}\n"
                + f"New process: cmd: {argv!r}\n"
                + f"New process parent pid: {os.getppid()!r}\n"
            )

        if welcome:
            self.did_welcome = True
            return welcome + text
        else:
            return text


class PytestTracker:
    """Track the current pytest test name to add to debug messages."""

    def __init__(self) -> None:
        self.test_name: str | None = None

    def filter(self, text: str) -> str:
        """Add a message when the pytest test changes."""
        test_name = os.getenv("PYTEST_CURRENT_TEST")
        if test_name != self.test_name:
            text = f"Pytest context: {test_name}\n{text}"
            self.test_name = test_name
        return text


class DebugOutputFile:
    """A file-like object that includes pid and cwd information."""

    def __init__(
        self,
        outfile: IO[str] | None,
        filters: Iterable[Callable[[str], str]],
    ):
        self.outfile = outfile
        self.filters = list(filters)
        self.pid = os.getpid()

    @classmethod
    def get_one(
        cls,
        fileobj: IO[str] | None = None,
        file_name: str | None = None,
        filters: Iterable[Callable[[str], str]] = (),
        interim: bool = False,
    ) -> DebugOutputFile:
        """Get a DebugOutputFile.

        If `fileobj` is provided, then a new DebugOutputFile is made with it.

        If `fileobj` isn't provided, then a file is chosen (`file_name` if
        provided, or COVERAGE_DEBUG_FILE, or stderr), and a process-wide
        singleton DebugOutputFile is made.

        `filters` are the text filters to apply to the stream to annotate with
        pids, etc.

        If `interim` is true, then a future `get_one` can replace this one.

        """
        if fileobj is not None:
            # Make DebugOutputFile around the fileobj passed.
            return cls(fileobj, filters)

        the_one, is_interim = cls._get_singleton_data()
        if the_one is None or is_interim:
            if file_name is not None:
                fileobj = open(file_name, "a", encoding="utf-8")
            else:
                # $set_env.py: COVERAGE_DEBUG_FILE - Where to write debug output
                file_name = os.getenv("COVERAGE_DEBUG_FILE", FORCED_DEBUG_FILE)
                if file_name in ["stdout", "stderr"]:
                    fileobj = getattr(sys, file_name)
                elif file_name:
                    fileobj = open(file_name, "a", encoding="utf-8")
                    atexit.register(fileobj.close)
                else:
                    fileobj = sys.stderr
            the_one = cls(fileobj, filters)
            cls._set_singleton_data(the_one, interim)

        if not (the_one.filters):
            the_one.filters = list(filters)
        return the_one

    # Because of the way igor.py deletes and re-imports modules,
    # this class can be defined more than once. But we really want
    # a process-wide singleton. So stash it in sys.modules instead of
    # on a class attribute. Yes, this is aggressively gross.

    SYS_MOD_NAME: Final[str] = "$coverage.debug.DebugOutputFile.the_one"
    SINGLETON_ATTR: Final[str] = "the_one_and_is_interim"

    @classmethod
    def _set_singleton_data(cls, the_one: DebugOutputFile, interim: bool) -> None:
        """Set the one DebugOutputFile to rule them all."""
        singleton_module = types.ModuleType(cls.SYS_MOD_NAME)
        setattr(singleton_module, cls.SINGLETON_ATTR, (the_one, interim))
        sys.modules[cls.SYS_MOD_NAME] = singleton_module

    @classmethod
    def _get_singleton_data(cls) -> tuple[DebugOutputFile | None, bool]:
        """Get the one DebugOutputFile."""
        singleton_module = sys.modules.get(cls.SYS_MOD_NAME)
        return getattr(singleton_module, cls.SINGLETON_ATTR, (None, True))

    @classmethod
    def _del_singleton_data(cls) -> None:
        """Delete the one DebugOutputFile, just for tests to use."""
        if cls.SYS_MOD_NAME in sys.modules:
            del sys.modules[cls.SYS_MOD_NAME]

    def write(self, text: str) -> None:
        """Just like file.write, but filter through all our filters."""
        assert self.outfile is not None
        if not self.outfile.closed:
            self.outfile.write(filter_text(text, self.filters))
            self.outfile.flush()

    def flush(self) -> None:
        """Flush our file."""
        assert self.outfile is not None
        if not self.outfile.closed:
            self.outfile.flush()


def log(msg: str, stack: bool = False) -> None:  # pragma: debugging
    """Write a log message as forcefully as possible."""
    out = DebugOutputFile.get_one(interim=True)
    out.write(msg + "\n")
    if stack:
        dump_stack_frames(out=out, skip=1)


def decorate_methods(
    decorator: Callable[..., Any],
    butnot: Iterable[str] = (),
    private: bool = False,
) -> Callable[..., Any]:  # pragma: debugging
    """A class decorator to apply a decorator to methods."""

    def _decorator(cls):  # type: ignore[no-untyped-def]
        for name, meth in inspect.getmembers(cls, inspect.isroutine):
            if name not in cls.__dict__:
                continue
            if name != "__init__":
                if not private and name.startswith("_"):
                    continue
            if name in butnot:
                continue
            setattr(cls, name, decorator(meth))
        return cls

    return _decorator


def break_in_pudb(func: AnyCallable) -> AnyCallable:  # pragma: debugging
    """A function decorator to stop in the debugger for each call."""

    @functools.wraps(func)
    def _wrapper(*args: Any, **kwargs: Any) -> Any:
        import pudb

        sys.stdout = sys.__stdout__
        pudb.set_trace()
        return func(*args, **kwargs)

    return _wrapper


OBJ_IDS = itertools.count()
CALLS = itertools.count()
OBJ_ID_ATTR = "$coverage.object_id"


def show_calls(
    show_args: bool = True,
    show_stack: bool = False,
    show_return: bool = False,
) -> Callable[..., Any]:  # pragma: debugging
    """A method decorator to debug-log each call to the function."""

    def _decorator(func: AnyCallable) -> AnyCallable:
        @functools.wraps(func)
        def _wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
            oid = getattr(self, OBJ_ID_ATTR, None)
            if oid is None:
                oid = f"{os.getpid():08d} {next(OBJ_IDS):04d}"
                setattr(self, OBJ_ID_ATTR, oid)
            extra = ""
            if show_args:
                eargs = ", ".join(map(repr, args))
                ekwargs = ", ".join("{}={!r}".format(*item) for item in kwargs.items())
                extra += "("
                extra += eargs
                if eargs and ekwargs:
                    extra += ", "
                extra += ekwargs
                extra += ")"
            if show_stack:
                extra += " @ "
                extra += "; ".join(short_stack(short_filenames=True).splitlines())
            callid = next(CALLS)
            msg = f"{oid} {callid:04d} {func.__name__}{extra}\n"
            DebugOutputFile.get_one(interim=True).write(msg)
            ret = func(self, *args, **kwargs)
            if show_return:
                msg = f"{oid} {callid:04d} {func.__name__} return {ret!r}\n"
                DebugOutputFile.get_one(interim=True).write(msg)
            return ret

        return _wrapper

    return _decorator


def relevant_environment_display(env: Mapping[str, str]) -> list[tuple[str, str]]:
    """Filter environment variables for a debug display.

    Select variables to display (with COV or PY in the name, or HOME, TEMP, or
    TMP), and also cloak sensitive values with asterisks.

    Arguments:
        env: a dict of environment variable names and values.

    Returns:
        A list of pairs (name, value) to show.

    """
    SLUGS = {"COV", "PY"}
    INCLUDE = {"HOME", "TEMP", "TMP"}
    CLOAK = {"API", "TOKEN", "KEY", "SECRET", "PASS", "SIGNATURE"}
    TRUNCATE = {"COVERAGE_PROCESS_CONFIG"}
    TRUNCATE_LEN = 60

    to_show = []
    for name, val in env.items():
        show = False
        if name in INCLUDE:
            show = True
        elif any(slug in name for slug in SLUGS):
            show = True
        if show:
            if any(slug in name for slug in CLOAK):
                val = re.sub(r"\w", "*", val)
            if name in TRUNCATE:
                if len(val) > TRUNCATE_LEN:
                    val = val[: TRUNCATE_LEN - 3] + "..."
            to_show.append((name, val))
    return human_sorted_items(to_show)