| # Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0 |
| # For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt |
| |
| """Support for plugins.""" |
| |
| from __future__ import annotations |
| |
| import os |
| import os.path |
| import sys |
| |
| from types import FrameType |
| from typing import Any, Iterable, Iterator |
| |
| from coverage.exceptions import PluginError |
| from coverage.misc import isolate_module |
| from coverage.plugin import CoveragePlugin, FileTracer, FileReporter |
| from coverage.types import ( |
| TArc, TConfigurable, TDebugCtl, TLineNo, TPluginConfig, TSourceTokenLines, |
| ) |
| |
| os = isolate_module(os) |
| |
| |
| class Plugins: |
| """The currently loaded collection of coverage.py plugins.""" |
| |
| def __init__(self) -> None: |
| self.order: list[CoveragePlugin] = [] |
| self.names: dict[str, CoveragePlugin] = {} |
| self.file_tracers: list[CoveragePlugin] = [] |
| self.configurers: list[CoveragePlugin] = [] |
| self.context_switchers: list[CoveragePlugin] = [] |
| |
| self.current_module: str | None = None |
| self.debug: TDebugCtl | None |
| |
| @classmethod |
| def load_plugins( |
| cls, |
| modules: Iterable[str], |
| config: TPluginConfig, |
| debug: TDebugCtl | None = None, |
| ) -> Plugins: |
| """Load plugins from `modules`. |
| |
| Returns a Plugins object with the loaded and configured plugins. |
| |
| """ |
| plugins = cls() |
| plugins.debug = debug |
| |
| for module in modules: |
| plugins.current_module = module |
| __import__(module) |
| mod = sys.modules[module] |
| |
| coverage_init = getattr(mod, "coverage_init", None) |
| if not coverage_init: |
| raise PluginError( |
| f"Plugin module {module!r} didn't define a coverage_init function", |
| ) |
| |
| options = config.get_plugin_options(module) |
| coverage_init(plugins, options) |
| |
| plugins.current_module = None |
| return plugins |
| |
| def add_file_tracer(self, plugin: CoveragePlugin) -> None: |
| """Add a file tracer plugin. |
| |
| `plugin` is an instance of a third-party plugin class. It must |
| implement the :meth:`CoveragePlugin.file_tracer` method. |
| |
| """ |
| self._add_plugin(plugin, self.file_tracers) |
| |
| def add_configurer(self, plugin: CoveragePlugin) -> None: |
| """Add a configuring plugin. |
| |
| `plugin` is an instance of a third-party plugin class. It must |
| implement the :meth:`CoveragePlugin.configure` method. |
| |
| """ |
| self._add_plugin(plugin, self.configurers) |
| |
| def add_dynamic_context(self, plugin: CoveragePlugin) -> None: |
| """Add a dynamic context plugin. |
| |
| `plugin` is an instance of a third-party plugin class. It must |
| implement the :meth:`CoveragePlugin.dynamic_context` method. |
| |
| """ |
| self._add_plugin(plugin, self.context_switchers) |
| |
| def add_noop(self, plugin: CoveragePlugin) -> None: |
| """Add a plugin that does nothing. |
| |
| This is only useful for testing the plugin support. |
| |
| """ |
| self._add_plugin(plugin, None) |
| |
| def _add_plugin( |
| self, |
| plugin: CoveragePlugin, |
| specialized: list[CoveragePlugin] | None, |
| ) -> None: |
| """Add a plugin object. |
| |
| `plugin` is a :class:`CoveragePlugin` instance to add. `specialized` |
| is a list to append the plugin to. |
| |
| """ |
| plugin_name = f"{self.current_module}.{plugin.__class__.__name__}" |
| if self.debug and self.debug.should("plugin"): |
| self.debug.write(f"Loaded plugin {self.current_module!r}: {plugin!r}") |
| labelled = LabelledDebug(f"plugin {self.current_module!r}", self.debug) |
| plugin = DebugPluginWrapper(plugin, labelled) |
| |
| plugin._coverage_plugin_name = plugin_name |
| plugin._coverage_enabled = True |
| self.order.append(plugin) |
| self.names[plugin_name] = plugin |
| if specialized is not None: |
| specialized.append(plugin) |
| |
| def __bool__(self) -> bool: |
| return bool(self.order) |
| |
| def __iter__(self) -> Iterator[CoveragePlugin]: |
| return iter(self.order) |
| |
| def get(self, plugin_name: str) -> CoveragePlugin: |
| """Return a plugin by name.""" |
| return self.names[plugin_name] |
| |
| |
| class LabelledDebug: |
| """A Debug writer, but with labels for prepending to the messages.""" |
| |
| def __init__(self, label: str, debug: TDebugCtl, prev_labels: Iterable[str] = ()): |
| self.labels = list(prev_labels) + [label] |
| self.debug = debug |
| |
| def add_label(self, label: str) -> LabelledDebug: |
| """Add a label to the writer, and return a new `LabelledDebug`.""" |
| return LabelledDebug(label, self.debug, self.labels) |
| |
| def message_prefix(self) -> str: |
| """The prefix to use on messages, combining the labels.""" |
| prefixes = self.labels + [""] |
| return ":\n".join(" "*i+label for i, label in enumerate(prefixes)) |
| |
| def write(self, message: str) -> None: |
| """Write `message`, but with the labels prepended.""" |
| self.debug.write(f"{self.message_prefix()}{message}") |
| |
| |
| class DebugPluginWrapper(CoveragePlugin): |
| """Wrap a plugin, and use debug to report on what it's doing.""" |
| |
| def __init__(self, plugin: CoveragePlugin, debug: LabelledDebug) -> None: |
| super().__init__() |
| self.plugin = plugin |
| self.debug = debug |
| |
| def file_tracer(self, filename: str) -> FileTracer | None: |
| tracer = self.plugin.file_tracer(filename) |
| self.debug.write(f"file_tracer({filename!r}) --> {tracer!r}") |
| if tracer: |
| debug = self.debug.add_label(f"file {filename!r}") |
| tracer = DebugFileTracerWrapper(tracer, debug) |
| return tracer |
| |
| def file_reporter(self, filename: str) -> FileReporter | str: |
| reporter = self.plugin.file_reporter(filename) |
| assert isinstance(reporter, FileReporter) |
| self.debug.write(f"file_reporter({filename!r}) --> {reporter!r}") |
| if reporter: |
| debug = self.debug.add_label(f"file {filename!r}") |
| reporter = DebugFileReporterWrapper(filename, reporter, debug) |
| return reporter |
| |
| def dynamic_context(self, frame: FrameType) -> str | None: |
| context = self.plugin.dynamic_context(frame) |
| self.debug.write(f"dynamic_context({frame!r}) --> {context!r}") |
| return context |
| |
| def find_executable_files(self, src_dir: str) -> Iterable[str]: |
| executable_files = self.plugin.find_executable_files(src_dir) |
| self.debug.write(f"find_executable_files({src_dir!r}) --> {executable_files!r}") |
| return executable_files |
| |
| def configure(self, config: TConfigurable) -> None: |
| self.debug.write(f"configure({config!r})") |
| self.plugin.configure(config) |
| |
| def sys_info(self) -> Iterable[tuple[str, Any]]: |
| return self.plugin.sys_info() |
| |
| |
| class DebugFileTracerWrapper(FileTracer): |
| """A debugging `FileTracer`.""" |
| |
| def __init__(self, tracer: FileTracer, debug: LabelledDebug) -> None: |
| self.tracer = tracer |
| self.debug = debug |
| |
| def _show_frame(self, frame: FrameType) -> str: |
| """A short string identifying a frame, for debug messages.""" |
| return "%s@%d" % ( |
| os.path.basename(frame.f_code.co_filename), |
| frame.f_lineno, |
| ) |
| |
| def source_filename(self) -> str: |
| sfilename = self.tracer.source_filename() |
| self.debug.write(f"source_filename() --> {sfilename!r}") |
| return sfilename |
| |
| def has_dynamic_source_filename(self) -> bool: |
| has = self.tracer.has_dynamic_source_filename() |
| self.debug.write(f"has_dynamic_source_filename() --> {has!r}") |
| return has |
| |
| def dynamic_source_filename(self, filename: str, frame: FrameType) -> str | None: |
| dyn = self.tracer.dynamic_source_filename(filename, frame) |
| self.debug.write("dynamic_source_filename({!r}, {}) --> {!r}".format( |
| filename, self._show_frame(frame), dyn, |
| )) |
| return dyn |
| |
| def line_number_range(self, frame: FrameType) -> tuple[TLineNo, TLineNo]: |
| pair = self.tracer.line_number_range(frame) |
| self.debug.write(f"line_number_range({self._show_frame(frame)}) --> {pair!r}") |
| return pair |
| |
| |
| class DebugFileReporterWrapper(FileReporter): |
| """A debugging `FileReporter`.""" |
| |
| def __init__(self, filename: str, reporter: FileReporter, debug: LabelledDebug) -> None: |
| super().__init__(filename) |
| self.reporter = reporter |
| self.debug = debug |
| |
| def relative_filename(self) -> str: |
| ret = self.reporter.relative_filename() |
| self.debug.write(f"relative_filename() --> {ret!r}") |
| return ret |
| |
| def lines(self) -> set[TLineNo]: |
| ret = self.reporter.lines() |
| self.debug.write(f"lines() --> {ret!r}") |
| return ret |
| |
| def excluded_lines(self) -> set[TLineNo]: |
| ret = self.reporter.excluded_lines() |
| self.debug.write(f"excluded_lines() --> {ret!r}") |
| return ret |
| |
| def translate_lines(self, lines: Iterable[TLineNo]) -> set[TLineNo]: |
| ret = self.reporter.translate_lines(lines) |
| self.debug.write(f"translate_lines({lines!r}) --> {ret!r}") |
| return ret |
| |
| def translate_arcs(self, arcs: Iterable[TArc]) -> set[TArc]: |
| ret = self.reporter.translate_arcs(arcs) |
| self.debug.write(f"translate_arcs({arcs!r}) --> {ret!r}") |
| return ret |
| |
| def no_branch_lines(self) -> set[TLineNo]: |
| ret = self.reporter.no_branch_lines() |
| self.debug.write(f"no_branch_lines() --> {ret!r}") |
| return ret |
| |
| def exit_counts(self) -> dict[TLineNo, int]: |
| ret = self.reporter.exit_counts() |
| self.debug.write(f"exit_counts() --> {ret!r}") |
| return ret |
| |
| def arcs(self) -> set[TArc]: |
| ret = self.reporter.arcs() |
| self.debug.write(f"arcs() --> {ret!r}") |
| return ret |
| |
| def source(self) -> str: |
| ret = self.reporter.source() |
| self.debug.write("source() --> %d chars" % (len(ret),)) |
| return ret |
| |
| def source_token_lines(self) -> TSourceTokenLines: |
| ret = list(self.reporter.source_token_lines()) |
| self.debug.write("source_token_lines() --> %d tokens" % (len(ret),)) |
| return ret |
