FM / coverage

/opt/cloudlinux/venv/lib/python3.11/site-packages/coverage/

/opt/cloudlinux/venv/lib/python3.11/site-packages/coverage

Name	Size	Mode	Actions
fullcoverage/	-	0755	rm
htmlfiles/	-	0755	rm
__pycache__/	-	0755	rm
annotate.py	3758	0644	edit dl rm
bytecode.py	713	0644	edit dl rm
cmdline.py	34427	0644	edit dl rm
collector.py	20558	0644	edit dl rm
config.py	21989	0644	edit dl rm
context.py	2483	0644	edit dl rm
control.py	51705	0644	edit dl rm
data.py	7508	0644	edit dl rm
debug.py	17805	0644	edit dl rm
disposition.py	1916	0644	edit dl rm
env.py	6081	0644	edit dl rm
exceptions.py	1362	0644	edit dl rm
execfile.py	12133	0644	edit dl rm
files.py	19374	0644	edit dl rm
html.py	23165	0644	edit dl rm
inorout.py	23900	0644	edit dl rm
jsonreport.py	4754	0644	edit dl rm
lcovreport.py	4939	0644	edit dl rm
misc.py	12172	0644	edit dl rm
multiproc.py	3846	0644	edit dl rm
numbits.py	4669	0644	edit dl rm
parser.py	56720	0644	edit dl rm
phystokens.py	7805	0644	edit dl rm
plugin.py	19525	0644	edit dl rm
plugin_support.py	10351	0644	edit dl rm
py.typed	72	0644	edit dl rm
python.py	8066	0644	edit dl rm
pytracer.py	14420	0644	edit dl rm
report.py	10623	0644	edit dl rm
report_core.py	4068	0644	edit dl rm
results.py	13384	0644	edit dl rm
sqldata.py	51295	0644	edit dl rm
templite.py	10952	0644	edit dl rm
tomlconfig.py	7569	0644	edit dl rm
tracer.cpython-311-x86_64-linux-gnu.so	110352	0755	edit dl rm
types.py	5526	0644	edit dl rm
version.py	1431	0644	edit dl rm
xmlreport.py	9795	0644	edit dl rm
__init__.py	1284	0644	edit dl rm
__main__.py	257	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/coverage/pytracer.py (14420B)

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

"""Raw data collector for coverage.py."""

from __future__ import annotations

import atexit
import dis
import sys
import threading

from types import FrameType, ModuleType
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, cast

from coverage import env
from coverage.types import (
    TArc, TFileDisposition, TLineNo, TTraceData, TTraceFileData, TTraceFn,
    TTracer, TWarnFn,
)

# We need the YIELD_VALUE opcode below, in a comparison-friendly form.
RESUME = dis.opmap.get("RESUME")
RETURN_VALUE = dis.opmap["RETURN_VALUE"]
if RESUME is None:
    YIELD_VALUE = dis.opmap["YIELD_VALUE"]
    YIELD_FROM = dis.opmap["YIELD_FROM"]
    YIELD_FROM_OFFSET = 0 if env.PYPY else 2

# When running meta-coverage, this file can try to trace itself, which confuses
# everything.  Don't trace ourselves.

THIS_FILE = __file__.rstrip("co")

class PyTracer(TTracer):
    """Python implementation of the raw data tracer."""

# Because of poor implementations of trace-function-manipulating tools,
    # the Python trace function must be kept very simple.  In particular, there
    # must be only one function ever set as the trace function, both through
    # sys.settrace, and as the return value from the trace function.  Put
    # another way, the trace function must always return itself.  It cannot
    # swap in other functions, or return None to avoid tracing a particular
    # frame.
    #
    # The trace manipulator that introduced this restriction is DecoratorTools,
    # which sets a trace function, and then later restores the pre-existing one
    # by calling sys.settrace with a function it found in the current frame.
    #
    # Systems that use DecoratorTools (or similar trace manipulations) must use
    # PyTracer to get accurate results.  The command-line --timid argument is
    # used to force the use of this tracer.

def __init__(self) -> None:
        # pylint: disable=super-init-not-called
        # Attributes set from the collector:
        self.data: TTraceData
        self.trace_arcs = False
        self.should_trace: Callable[[str, FrameType], TFileDisposition]
        self.should_trace_cache: Dict[str, Optional[TFileDisposition]]
        self.should_start_context: Optional[Callable[[FrameType], Optional[str]]] = None
        self.switch_context: Optional[Callable[[Optional[str]], None]] = None
        self.warn: TWarnFn

# The threading module to use, if any.
        self.threading: Optional[ModuleType] = None

self.cur_file_data: Optional[TTraceFileData] = None
        self.last_line: TLineNo = 0
        self.cur_file_name: Optional[str] = None
        self.context: Optional[str] = None
        self.started_context = False

self.data_stack: List[Tuple[Optional[TTraceFileData], Optional[str], TLineNo, bool]] = []
        self.thread: Optional[threading.Thread] = None
        self.stopped = False
        self._activity = False

self.in_atexit = False
        # On exit, self.in_atexit = True
        atexit.register(setattr, self, "in_atexit", True)

# Cache a bound method on the instance, so that we don't have to
        # re-create a bound method object all the time.
        self._cached_bound_method_trace: TTraceFn = self._trace

def __repr__(self) -> str:
        me = id(self)
        points = sum(len(v) for v in self.data.values())
        files = len(self.data)
        return f"<PyTracer at 0x{me:x}: {points} data points in {files} files>"

def log(self, marker: str, *args: Any) -> None:
        """For hard-core logging of what this tracer is doing."""
        with open("/tmp/debug_trace.txt", "a") as f:
            f.write("{} {}[{}]".format(
                marker,
                id(self),
                len(self.data_stack),
            ))
            if 0:   # if you want thread ids..
                f.write(".{:x}.{:x}".format(                    # type: ignore[unreachable]
                    self.thread.ident,
                    self.threading.current_thread().ident,
                ))
            f.write(" {}".format(" ".join(map(str, args))))
            if 0:   # if you want callers..
                f.write(" | ")                                  # type: ignore[unreachable]
                stack = " / ".join(
                    (fname or "???").rpartition("/")[-1]
                    for _, fname, _, _ in self.data_stack
                )
                f.write(stack)
            f.write("\n")

def _trace(
        self,
        frame: FrameType,
        event: str,
        arg: Any,                               # pylint: disable=unused-argument
        lineno: Optional[TLineNo] = None,       # pylint: disable=unused-argument
    ) -> Optional[TTraceFn]:
        """The trace function passed to sys.settrace."""

if THIS_FILE in frame.f_code.co_filename:
            return None

#self.log(":", frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name + "()", event)

if (self.stopped and sys.gettrace() == self._cached_bound_method_trace):    # pylint: disable=comparison-with-callable
            # The PyTrace.stop() method has been called, possibly by another
            # thread, let's deactivate ourselves now.
            if 0:
                f = frame                           # type: ignore[unreachable]
                self.log("---\nX", f.f_code.co_filename, f.f_lineno)
                while f:
                    self.log(">", f.f_code.co_filename, f.f_lineno, f.f_code.co_name, f.f_trace)
                    f = f.f_back
            sys.settrace(None)
            try:
                self.cur_file_data, self.cur_file_name, self.last_line, self.started_context = (
                    self.data_stack.pop()
                )
            except IndexError:
                self.log(
                    "Empty stack!",
                    frame.f_code.co_filename,
                    frame.f_lineno,
                    frame.f_code.co_name
                )
            return None

# if event != "call" and frame.f_code.co_filename != self.cur_file_name:
        #     self.log("---\n*", frame.f_code.co_filename, self.cur_file_name, frame.f_lineno)

if event == "call":
            # Should we start a new context?
            if self.should_start_context and self.context is None:
                context_maybe = self.should_start_context(frame)
                if context_maybe is not None:
                    self.context = context_maybe
                    started_context = True
                    assert self.switch_context is not None
                    self.switch_context(self.context)
                else:
                    started_context = False
            else:
                started_context = False
            self.started_context = started_context

# Entering a new frame.  Decide if we should trace in this file.
            self._activity = True
            self.data_stack.append(
                (
                    self.cur_file_data,
                    self.cur_file_name,
                    self.last_line,
                    started_context,
                )
            )

# Improve tracing performance: when calling a function, both caller
            # and callee are often within the same file. if that's the case, we
            # don't have to re-check whether to trace the corresponding
            # function (which is a little bit expensive since it involves
            # dictionary lookups). This optimization is only correct if we
            # didn't start a context.
            filename = frame.f_code.co_filename
            if filename != self.cur_file_name or started_context:
                self.cur_file_name = filename
                disp = self.should_trace_cache.get(filename)
                if disp is None:
                    disp = self.should_trace(filename, frame)
                    self.should_trace_cache[filename] = disp

self.cur_file_data = None
                if disp.trace:
                    tracename = disp.source_filename
                    assert tracename is not None
                    if tracename not in self.data:
                        self.data[tracename] = set()    # type: ignore[assignment]
                    self.cur_file_data = self.data[tracename]
                else:
                    frame.f_trace_lines = False
            elif not self.cur_file_data:
                frame.f_trace_lines = False

# The call event is really a "start frame" event, and happens for
            # function calls and re-entering generators.  The f_lasti field is
            # -1 for calls, and a real offset for generators.  Use <0 as the
            # line number for calls, and the real line number for generators.
            if RESUME is not None:
                # The current opcode is guaranteed to be RESUME. The argument
                # determines what kind of resume it is.
                oparg = frame.f_code.co_code[frame.f_lasti + 1]
                real_call = (oparg == 0)
            else:
                real_call = (getattr(frame, "f_lasti", -1) < 0)
            if real_call:
                self.last_line = -frame.f_code.co_firstlineno
            else:
                self.last_line = frame.f_lineno

elif event == "line":
            # Record an executed line.
            if self.cur_file_data is not None:
                flineno: TLineNo = frame.f_lineno

if self.trace_arcs:
                    cast(Set[TArc], self.cur_file_data).add((self.last_line, flineno))
                else:
                    cast(Set[TLineNo], self.cur_file_data).add(flineno)
                self.last_line = flineno

elif event == "return":
            if self.trace_arcs and self.cur_file_data:
                # Record an arc leaving the function, but beware that a
                # "return" event might just mean yielding from a generator.
                code = frame.f_code.co_code
                lasti = frame.f_lasti
                if RESUME is not None:
                    if len(code) == lasti + 2:
                        # A return from the end of a code object is a real return.
                        real_return = True
                    else:
                        # it's a real return.
                        real_return = (code[lasti + 2] != RESUME)
                else:
                    if code[lasti] == RETURN_VALUE:
                        real_return = True
                    elif code[lasti] == YIELD_VALUE:
                        real_return = False
                    elif len(code) <= lasti + YIELD_FROM_OFFSET:
                        real_return = True
                    elif code[lasti + YIELD_FROM_OFFSET] == YIELD_FROM:
                        real_return = False
                    else:
                        real_return = True
                if real_return:
                    first = frame.f_code.co_firstlineno
                    cast(Set[TArc], self.cur_file_data).add((self.last_line, -first))

# Leaving this function, pop the filename stack.
            self.cur_file_data, self.cur_file_name, self.last_line, self.started_context = (
                self.data_stack.pop()
            )
            # Leaving a context?
            if self.started_context:
                assert self.switch_context is not None
                self.context = None
                self.switch_context(None)
        return self._cached_bound_method_trace

def start(self) -> TTraceFn:
        """Start this Tracer.

Return a Python function suitable for use with sys.settrace().

"""
        self.stopped = False
        if self.threading:
            if self.thread is None:
                self.thread = self.threading.current_thread()
            else:
                if self.thread.ident != self.threading.current_thread().ident:
                    # Re-starting from a different thread!? Don't set the trace
                    # function, but we are marked as running again, so maybe it
                    # will be ok?
                    #self.log("~", "starting on different threads")
                    return self._cached_bound_method_trace

sys.settrace(self._cached_bound_method_trace)
        return self._cached_bound_method_trace

def stop(self) -> None:
        """Stop this Tracer."""
        # Get the active tracer callback before setting the stop flag to be
        # able to detect if the tracer was changed prior to stopping it.
        tf = sys.gettrace()

# Set the stop flag. The actual call to sys.settrace(None) will happen
        # in the self._trace callback itself to make sure to call it from the
        # right thread.
        self.stopped = True

if self.threading:
            assert self.thread is not None
            if self.thread.ident != self.threading.current_thread().ident:
                # Called on a different thread than started us: we can't unhook
                # ourselves, but we've set the flag that we should stop, so we
                # won't do any more tracing.
                #self.log("~", "stopping on different threads")
                return

if self.warn:
            # PyPy clears the trace function before running atexit functions,
            # so don't warn if we are in atexit on PyPy and the trace function
            # has changed to None.
            dont_warn = (env.PYPY and env.PYPYVERSION >= (5, 4) and self.in_atexit and tf is None)
            if (not dont_warn) and tf != self._cached_bound_method_trace:   # pylint: disable=comparison-with-callable
                self.warn(
                    "Trace function changed, data is likely wrong: " +
                    f"{tf!r} != {self._cached_bound_method_trace!r}",
                    slug="trace-changed",
                )

def activity(self) -> bool:
        """Has there been any activity?"""
        return self._activity

def reset_activity(self) -> None:
        """Reset the activity() flag."""
        self._activity = False

def get_stats(self) -> Optional[Dict[str, int]]:
        """Return a dictionary of statistics, or None."""
        return None

cmd: