From 8ed626a189628cc154cc1707b544b3361220d46d Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:40:36 +0300 Subject: [PATCH 01/15] plugin: replace coverage.X with from coverage import X The Coverage and CoverageData classes were using using full paths while other classes were imported. For consistency just import all classes. --- coverage_sh/plugin.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index aec14b1..973f0ca 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -20,10 +20,9 @@ from typing import TYPE_CHECKING, Any, cast from warnings import warn -import coverage import magic import tree_sitter_bash -from coverage import CoveragePlugin, FileReporter, FileTracer +from coverage import Coverage, CoverageData, CoveragePlugin, FileReporter, FileTracer from tree_sitter import Language, Parser if TYPE_CHECKING: @@ -163,7 +162,7 @@ def __init__(self, coverage_data_path: Path): def write(self, line_data: LineData) -> None: suffix_ = "sh." + filename_suffix() - coverage_data = coverage.CoverageData( + coverage_data = CoverageData( basename=self._coverage_data_path, suffix=suffix_, # TODO: set warn, debug and no_disk @@ -256,7 +255,8 @@ class PatchedPopen(OriginalPopen): # type: ignore[type-arg] data_file_path: Path = Path.cwd() def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] - if coverage.Coverage.current() is None: + curcov = Coverage.current() + if curcov is None: # we are not recording coverage, so just act like the original Popen self._parser_thread = None super().__init__(*args, **kwargs) From 94314fe24f508e291787b81be46d42b0a15863c9 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:32:55 +0300 Subject: [PATCH 02/15] CoverageParserThread: initial DebugControl support --- coverage_sh/plugin.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 973f0ca..6c57446 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -23,6 +23,7 @@ import magic import tree_sitter_bash from coverage import Coverage, CoverageData, CoveragePlugin, FileReporter, FileTracer +from coverage.debug import DebugControl, NoDebugging from tree_sitter import Language, Parser if TYPE_CHECKING: @@ -181,24 +182,33 @@ def __init__( coverage_writer: CoverageWriter, name: str | None = None, parser: CovLineParser | None = None, + debug: DebugControl | None = None, ) -> None: super().__init__(name=name) self._keep_running = True self._listening = False self._parser = parser or CovLineParser() self._coverage_writer = coverage_writer + self._debug = debug or NoDebugging() self.fifo_path = TMP_PATH / f"coverage-sh.{filename_suffix()}.pipe" with contextlib.suppress(FileNotFoundError): self.fifo_path.unlink() os.mkfifo(self.fifo_path, mode=stat.S_IRUSR | stat.S_IWUSR) + self._debug_write(f"init done fifo_path={self.fifo_path}") + + def _debug_write(self, msg: str) -> None: + if self._debug.should("shell-helper-thread"): + self._debug.write("CoverageParserThread: " + msg) def start(self) -> None: + self._debug_write("start") super().start() while not self._listening: sleep(0.0001) def stop(self) -> None: + self._debug_write("stop") self._keep_running = False def run(self) -> None: @@ -214,6 +224,8 @@ def run(self) -> None: data_incoming = True while not eof and (data_incoming or self._keep_running): events = sel.select(timeout=1) + if not len(events): + self._debug_write("select timeout, retry ...") data_incoming = len(events) > 0 for key, _ in events: buf = os.read(key.fd, 2**10) From 1b1aff44dab7306139020bb155f1e9d912267c4f Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:35:21 +0300 Subject: [PATCH 03/15] add get_coverage_debug helper --- coverage_sh/plugin.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 6c57446..64558b6 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -261,6 +261,16 @@ def init_helper(fifo_path: Path) -> Path: return helper_path +def get_coverage_debug(coverage: Coverage) -> DebugControl: + """ + Get the DebugControl instance from the main Coverage object + + This is not exposed through the public API so we this custom helper will + read a private attribute + """ + return coverage._debug # noqa: SLF001 + + # the proper way to do this would be using OriginalPopen[Any] but that is not supported by python 3.8, so we jusrt # ignore this for the time being class PatchedPopen(OriginalPopen): # type: ignore[type-arg] From d510001a75f8b949d1865e1f47eac0d04a4f9824 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:35:09 +0300 Subject: [PATCH 04/15] ShellPlugin: initial DebugControl support --- coverage_sh/plugin.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 64558b6..c443c3b 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -345,11 +345,18 @@ class ShellPlugin(CoveragePlugin): def __init__(self, options: dict[str, Any]): self.options = options self._helper_path: None | Path = None + self._debug: DebugControl = NoDebugging() def configure(self, config: TConfigurable) -> None: data_file_option = config.get_option("run:data_file") coverage_data_path = Path(cast("str", data_file_option)).absolute() + current_coverage = Coverage.current() + if current_coverage is not None: + self._debug = get_coverage_debug(current_coverage) + if self._debug.should("config"): + self._debug.write("ShellPlugin.configure") + if config.get_option("run:core") == "sysmon" or ( sys.version_info >= (3, 14) and config.get_option("run:core") is None ): @@ -364,6 +371,7 @@ def configure(self, config: TConfigurable) -> None: parser_thread = CoverageParserThread( coverage_writer=CoverageWriter(coverage_data_path), name=f"CoverageParserThread({coverage_data_path!s})", + debug=self._debug, ) parser_thread.start() From bf4afccc1e578ac864f8eb17ee5413260470c198 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:38:23 +0300 Subject: [PATCH 05/15] PatchedPopen: initial DebugControl support --- coverage_sh/plugin.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index c443c3b..17acdf1 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -275,15 +275,25 @@ def get_coverage_debug(coverage: Coverage) -> DebugControl: # ignore this for the time being class PatchedPopen(OriginalPopen): # type: ignore[type-arg] data_file_path: Path = Path.cwd() + _debug: DebugControl def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] curcov = Coverage.current() if curcov is None: # we are not recording coverage, so just act like the original Popen self._parser_thread = None + self._debug = NoDebugging() super().__init__(*args, **kwargs) return + # minimal init for __repr__ to work for COVERAGE_DEBUG=self + self.returncode = None + self.args = args + + # initialize debug control + self._debug = debug = get_coverage_debug(curcov) + self._debug_write("__init__") + # convert args into kwargs sig = inspect.signature(subprocess.Popen) kwargs.update(dict(zip(sig.parameters.keys(), args))) @@ -291,6 +301,7 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] self._parser_thread = CoverageParserThread( coverage_writer=CoverageWriter(coverage_data_path=self.data_file_path), name="CoverageParserThread(None)", + debug=debug, ) self._parser_thread.start() @@ -304,7 +315,9 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] super().__init__(**kwargs) def wait(self, timeout: float | None = None) -> int: + self._debug_write(f"wait timeout={timeout}") retval = super().wait(timeout) + self._debug_write(f"wait result={retval}") if self._parser_thread is None: # no coverage recording was active during __init__ return retval @@ -315,6 +328,10 @@ def wait(self, timeout: float | None = None) -> int: self._helper_path.unlink() return retval + def _debug_write(self, msg: str) -> None: + if self._debug.should("patch"): + self._debug.write("PatchedPopen: " + msg) + class MonitorThread(threading.Thread): def __init__( From 336b0a93c6e0b35734a8458c63f3772c43b80bf8 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:29:23 +0300 Subject: [PATCH 06/15] README.md: indent caveats/cover-always-mode under Usage heading --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f2349d4..e221eac 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,7 @@ The resulting coverage is then displayed alongside the coverage of the python fi ![coverage.sh report screenshot](doc/media/screenshot_html-report.png) -## Caveats +### Caveats The plugin works by patching the `subprocess.Popen` class to set the "ENV" and "BASH_ENV" environment variables before execution, to source a helper script which enables tracing. This approach comes with a few caveats: @@ -53,7 +53,7 @@ execution, to source a helper script which enables tracing. This approach comes - It will only cover shell scripts that are executed via the subprocess module. - Only bash and sh are supported -## Cover-Always Mode +### Cover-Always Mode When using the subprocess modue is not an option, coverage-sh can operate in "cover-always-mode", which is activated by setting From 33994b0e865dc20070b8e7e8043fef36696c19ff Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 12:29:38 +0300 Subject: [PATCH 07/15] README.md: document debug options --- README.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/README.md b/README.md index e221eac..e6a08e8 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,15 @@ starting pytest from coverage , e.g.: coverage run -m pytest arg1 arg2 arg3 ``` +## Debug Options + +The coverage-sh plugin uses coveragepy debug infrastructure. You can enable debug by setting the `COVERAGE_DEBUG` variable or by running coverage with the `--debug` flag. The following debug options apply to coverage-sh: + +* `patch`: logs the patching of subprocess.Popen. +* `shell-helper-thread`: logs events related to background threads created by coverage_sh + +More options are documented in the [coveragepy documentation](https://coverage.readthedocs.io/en/latest/commands/cmd_debug.html#debug-option). + ## License Licensed under the [MIT License](LICENSE.txt). From fb27d6bbd832d575f8e1de01248f9c45295fbc79 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 14:14:19 +0300 Subject: [PATCH 08/15] test_plugin: add DebugControlString helper, borrowed from coveragepy --- tests/test_plugin.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 6c9ac78..fc9a9a1 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,5 +1,6 @@ # SPDX-License-Identifier: MIT # Copyright (c) 2023-2024 Kilian Lackhove +import io import json import os import re @@ -7,6 +8,7 @@ import sys import threading from collections import defaultdict +from collections.abc import Iterable from importlib.metadata import version from pathlib import Path from socket import gethostname @@ -16,6 +18,7 @@ import coverage import pytest from coverage.config import CoverageConfig +from coverage.debug import DebugControl from packaging.version import Version from coverage_sh.plugin import ( @@ -144,6 +147,18 @@ INNER_PY_EXECUTED_LINES = [2] +class DebugControlString(DebugControl): + """A `DebugControl` that writes to a StringIO, for testing.""" + + def __init__(self, options: Iterable[str]) -> None: + self.io = io.StringIO() + super().__init__(options, self.io) + + def get_output(self) -> str: + """Get the output text from the `DebugControl`.""" + return self.io.getvalue() + + @pytest.fixture def examples_dir(resources_dir: Path) -> Path: return resources_dir / "examples" From 8cea4728b699be80911046efa01b0689996f01f3 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 14:14:43 +0300 Subject: [PATCH 09/15] TestCoverageParserThread: add test_start_stop_debug --- tests/test_plugin.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index fc9a9a1..3827cb6 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -14,6 +14,7 @@ from socket import gethostname from time import sleep from typing import cast +from unittest.mock import MagicMock import coverage import pytest @@ -422,6 +423,19 @@ def test_lines_should_match_reference(self) -> None: for filename, lines in COVERAGE_LINE_COVERAGE.items(): assert writer.line_data[filename] == lines + def test_start_stop_debug(self) -> None: + debug = DebugControlString(["shell-helper-thread"]) + parser_thread = CoverageParserThread( + coverage_writer=MagicMock(CoverageWriter), + debug=debug, + ) + parser_thread.start() + parser_thread.stop() + parser_thread.join() + debug_output = debug.get_output() + assert "CoverageParserThread: start" in debug_output + assert "CoverageParserThread: stop" in debug_output + class TestCoverageWriter: def test_write_should_produce_readable_file(self, dummy_project_dir: Path) -> None: From 8f36bc1a04f6c4283617d25fc258bca31e49a7e5 Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 14:41:18 +0300 Subject: [PATCH 10/15] TestShellPlugin: add test_mock_configure_cover_always_debug --- tests/test_plugin.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 3827cb6..c3af043 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -14,6 +14,7 @@ from socket import gethostname from time import sleep from typing import cast +from unittest import mock from unittest.mock import MagicMock import coverage @@ -606,3 +607,26 @@ def test_configure_should_set_bash_env_when_cover_always( config = CoverageConfig() plugin.configure(config) assert os.getenv("BASH_ENV") + + def test_mock_configure_cover_always_debug( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.delenv("BASH_ENV", raising=False) + plugin = ShellPlugin({"cover_always": True}) + debug = DebugControlString(["config"]) + plugin._debug = debug # noqa: SLF001 + with ( + mock.patch.object(coverage.Coverage, "current", return_value=None), + mock.patch("coverage_sh.plugin.CoverageParserThread") as mock_parser_thread, + mock.patch("coverage_sh.plugin.CoverageWriter"), + mock.patch("coverage_sh.plugin.MonitorThread"), + ): + config = CoverageConfig() + plugin.configure(config) + debug_output = debug.get_output() + # check DebugControl writes + assert "ShellPlugin.configure" in debug_output + assert mock_parser_thread.call_count == 1 + # check DebugControl is passed to CoverageParserThread + assert mock_parser_thread.call_args.kwargs["debug"] == debug From 50fe549c9b89d20803b8db084d418f749ab9f6ae Mon Sep 17 00:00:00 2001 From: Leonard Crestez Date: Tue, 31 Mar 2026 14:48:10 +0300 Subject: [PATCH 11/15] TestPatchedPopen: add test_debug_control --- tests/test_plugin.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index c3af043..f31e058 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -517,6 +517,23 @@ def test_call_should_execute_example( assert proc.stdout is not None assert proc.stdout.read() == END2END_STDOUT + def test_debug_control(self) -> None: + debug = DebugControlString(["patch"]) + mock_coverage = MagicMock(coverage.Coverage) + mock_coverage._debug = debug # noqa: SLF001 + with ( + mock.patch.object(coverage.Coverage, "current", return_value=mock_coverage), + ): + proc = PatchedPopen(["echo", "hello"], stdout=subprocess.PIPE) + out, err = proc.communicate() + assert out == b"hello\n" + assert err is None + assert proc.returncode == 0 + debug_output = debug.get_output() + assert "PatchedPopen: __init__" in debug_output + assert "PatchedPopen: wait timeout" in debug_output + assert "PatchedPopen: wait result" in debug_output + class TestMonitorThread: class MainThreadStub: From 9b7c202edc23b0950de10086015de863479a0fb5 Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Wed, 1 Apr 2026 22:17:10 +0200 Subject: [PATCH 12/15] alternative implementation --- README.md | 7 +--- coverage_sh/plugin.py | 78 ++++++++++++++++------------------- tests/test_plugin.py | 96 ++++++++++++++++++------------------------- 3 files changed, 77 insertions(+), 104 deletions(-) diff --git a/README.md b/README.md index e6a08e8..75e5fa3 100644 --- a/README.md +++ b/README.md @@ -74,12 +74,7 @@ coverage run -m pytest arg1 arg2 arg3 ## Debug Options -The coverage-sh plugin uses coveragepy debug infrastructure. You can enable debug by setting the `COVERAGE_DEBUG` variable or by running coverage with the `--debug` flag. The following debug options apply to coverage-sh: - -* `patch`: logs the patching of subprocess.Popen. -* `shell-helper-thread`: logs events related to background threads created by coverage_sh - -More options are documented in the [coveragepy documentation](https://coverage.readthedocs.io/en/latest/commands/cmd_debug.html#debug-option). +The coverage-sh plugin uses coveragepy debug infrastructure. You can enable debug by setting the `COVERAGE_DEBUG` variable or by running coverage with the `--debug` flag. Logging in coverage-sh is enabled by the `shell` option. More options are documented in the [coveragepy documentation](https://coverage.readthedocs.io/en/latest/commands/cmd_debug.html#debug-option). ## License diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 17acdf1..57fbdb4 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -23,12 +23,12 @@ import magic import tree_sitter_bash from coverage import Coverage, CoverageData, CoveragePlugin, FileReporter, FileTracer -from coverage.debug import DebugControl, NoDebugging from tree_sitter import Language, Parser if TYPE_CHECKING: from collections.abc import Iterable, Iterator + from coverage.debug import DebugControl from coverage.types import TConfigurable, TLineNo from tree_sitter import Node @@ -56,6 +56,28 @@ } SUPPORTED_MIME_TYPES = {"text/x-shellscript"} +PLUGIN_DEBUG_OPTION = "shell" + + +def debug_write(msg: str, debug_control: DebugControl | None = None) -> None: + + current_coverage = Coverage.current() + if current_coverage is None and debug_control is None: + # we are not recording coverage, so we have nowhere to send the message + return + + # DebugControl.write expects to be called from a frame with a "self" variable, so + # we use the same code to fetch that and pass it down to emulate that behavior + self = inspect.stack()[1][0].f_locals.get("self") # noqa: F841 + + try: + debug_control = debug_control or Coverage.current()._debug # type: ignore[union-attr] # noqa: SLF001 + + if debug_control.should(PLUGIN_DEBUG_OPTION): + debug_control.write(msg) + except Exception as e: # noqa: BLE001 + warn(f'Failed to log debug message: "{msg}": {e}', stacklevel=2) + class ShellFileReporter(FileReporter): def __init__(self, filename: str) -> None: @@ -182,33 +204,29 @@ def __init__( coverage_writer: CoverageWriter, name: str | None = None, parser: CovLineParser | None = None, - debug: DebugControl | None = None, ) -> None: super().__init__(name=name) self._keep_running = True self._listening = False self._parser = parser or CovLineParser() self._coverage_writer = coverage_writer - self._debug = debug or NoDebugging() self.fifo_path = TMP_PATH / f"coverage-sh.{filename_suffix()}.pipe" with contextlib.suppress(FileNotFoundError): self.fifo_path.unlink() os.mkfifo(self.fifo_path, mode=stat.S_IRUSR | stat.S_IWUSR) - self._debug_write(f"init done fifo_path={self.fifo_path}") - - def _debug_write(self, msg: str) -> None: - if self._debug.should("shell-helper-thread"): - self._debug.write("CoverageParserThread: " + msg) + debug_write( + f"init done fifo_path={self.fifo_path}", + ) def start(self) -> None: - self._debug_write("start") + debug_write("start") super().start() while not self._listening: sleep(0.0001) def stop(self) -> None: - self._debug_write("stop") + debug_write("stop") self._keep_running = False def run(self) -> None: @@ -225,7 +243,9 @@ def run(self) -> None: while not eof and (data_incoming or self._keep_running): events = sel.select(timeout=1) if not len(events): - self._debug_write("select timeout, retry ...") + debug_write( + "select timeout, retry ...", + ) data_incoming = len(events) > 0 for key, _ in events: buf = os.read(key.fd, 2**10) @@ -261,28 +281,15 @@ def init_helper(fifo_path: Path) -> Path: return helper_path -def get_coverage_debug(coverage: Coverage) -> DebugControl: - """ - Get the DebugControl instance from the main Coverage object - - This is not exposed through the public API so we this custom helper will - read a private attribute - """ - return coverage._debug # noqa: SLF001 - - # the proper way to do this would be using OriginalPopen[Any] but that is not supported by python 3.8, so we jusrt # ignore this for the time being class PatchedPopen(OriginalPopen): # type: ignore[type-arg] data_file_path: Path = Path.cwd() - _debug: DebugControl def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] - curcov = Coverage.current() - if curcov is None: + if Coverage.current() is None: # we are not recording coverage, so just act like the original Popen self._parser_thread = None - self._debug = NoDebugging() super().__init__(*args, **kwargs) return @@ -290,9 +297,7 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] self.returncode = None self.args = args - # initialize debug control - self._debug = debug = get_coverage_debug(curcov) - self._debug_write("__init__") + debug_write("__init__") # convert args into kwargs sig = inspect.signature(subprocess.Popen) @@ -301,7 +306,6 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] self._parser_thread = CoverageParserThread( coverage_writer=CoverageWriter(coverage_data_path=self.data_file_path), name="CoverageParserThread(None)", - debug=debug, ) self._parser_thread.start() @@ -315,9 +319,9 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] super().__init__(**kwargs) def wait(self, timeout: float | None = None) -> int: - self._debug_write(f"wait timeout={timeout}") + debug_write(f"wait timeout={timeout}") retval = super().wait(timeout) - self._debug_write(f"wait result={retval}") + debug_write(f"wait result={retval}") if self._parser_thread is None: # no coverage recording was active during __init__ return retval @@ -328,10 +332,6 @@ def wait(self, timeout: float | None = None) -> int: self._helper_path.unlink() return retval - def _debug_write(self, msg: str) -> None: - if self._debug.should("patch"): - self._debug.write("PatchedPopen: " + msg) - class MonitorThread(threading.Thread): def __init__( @@ -362,18 +362,11 @@ class ShellPlugin(CoveragePlugin): def __init__(self, options: dict[str, Any]): self.options = options self._helper_path: None | Path = None - self._debug: DebugControl = NoDebugging() def configure(self, config: TConfigurable) -> None: data_file_option = config.get_option("run:data_file") coverage_data_path = Path(cast("str", data_file_option)).absolute() - current_coverage = Coverage.current() - if current_coverage is not None: - self._debug = get_coverage_debug(current_coverage) - if self._debug.should("config"): - self._debug.write("ShellPlugin.configure") - if config.get_option("run:core") == "sysmon" or ( sys.version_info >= (3, 14) and config.get_option("run:core") is None ): @@ -388,7 +381,6 @@ def configure(self, config: TConfigurable) -> None: parser_thread = CoverageParserThread( coverage_writer=CoverageWriter(coverage_data_path), name=f"CoverageParserThread({coverage_data_path!s})", - debug=self._debug, ) parser_thread.start() diff --git a/tests/test_plugin.py b/tests/test_plugin.py index f31e058..360b2c0 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -14,8 +14,6 @@ from socket import gethostname from time import sleep from typing import cast -from unittest import mock -from unittest.mock import MagicMock import coverage import pytest @@ -32,6 +30,7 @@ PatchedPopen, ShellFileReporter, ShellPlugin, + debug_write, filename_suffix, ) @@ -237,6 +236,46 @@ def test_end2end( ) +class TestDebugWrite: + def test_should_not_log_when_dsabled(self) -> None: + debug_control = DebugControlString([]) + # we need a variable named `self` in the caller frame that is of a known type + self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 + + debug_write("foo", debug_control) + + assert debug_control.get_output() == "" + + def test_should_log_when_enabled(self) -> None: + debug_control = DebugControlString(["shell"]) + # we need a variable named `self` in the caller frame that is of a known type + self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 + + debug_write("foo", debug_control) + + assert debug_control.get_output() == "foo\n" + + def test_should_log_self_when_enabled(self) -> None: + debug_control = DebugControlString(["self", "shell"]) + # we need a variable named `self` in the caller frame that is of a known type + self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 + + debug_write("foo", debug_control) + + assert ( + "self: None: + debug_control = DebugControlString(["self", "callers", "shell"]) + # we need a variable named `self` in the caller frame that is of a known type + self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 + + debug_write("foo", debug_control) + + assert "test_should_log_callers_when_enabled" in debug_control.get_output() + + @pytest.fixture(scope="session") def covpy_installs_pth_at_install_time() -> None: """Skip if coveragepy does not install a .pth file into site-packages at install time. @@ -424,19 +463,6 @@ def test_lines_should_match_reference(self) -> None: for filename, lines in COVERAGE_LINE_COVERAGE.items(): assert writer.line_data[filename] == lines - def test_start_stop_debug(self) -> None: - debug = DebugControlString(["shell-helper-thread"]) - parser_thread = CoverageParserThread( - coverage_writer=MagicMock(CoverageWriter), - debug=debug, - ) - parser_thread.start() - parser_thread.stop() - parser_thread.join() - debug_output = debug.get_output() - assert "CoverageParserThread: start" in debug_output - assert "CoverageParserThread: stop" in debug_output - class TestCoverageWriter: def test_write_should_produce_readable_file(self, dummy_project_dir: Path) -> None: @@ -517,23 +543,6 @@ def test_call_should_execute_example( assert proc.stdout is not None assert proc.stdout.read() == END2END_STDOUT - def test_debug_control(self) -> None: - debug = DebugControlString(["patch"]) - mock_coverage = MagicMock(coverage.Coverage) - mock_coverage._debug = debug # noqa: SLF001 - with ( - mock.patch.object(coverage.Coverage, "current", return_value=mock_coverage), - ): - proc = PatchedPopen(["echo", "hello"], stdout=subprocess.PIPE) - out, err = proc.communicate() - assert out == b"hello\n" - assert err is None - assert proc.returncode == 0 - debug_output = debug.get_output() - assert "PatchedPopen: __init__" in debug_output - assert "PatchedPopen: wait timeout" in debug_output - assert "PatchedPopen: wait result" in debug_output - class TestMonitorThread: class MainThreadStub: @@ -624,26 +633,3 @@ def test_configure_should_set_bash_env_when_cover_always( config = CoverageConfig() plugin.configure(config) assert os.getenv("BASH_ENV") - - def test_mock_configure_cover_always_debug( - self, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - monkeypatch.delenv("BASH_ENV", raising=False) - plugin = ShellPlugin({"cover_always": True}) - debug = DebugControlString(["config"]) - plugin._debug = debug # noqa: SLF001 - with ( - mock.patch.object(coverage.Coverage, "current", return_value=None), - mock.patch("coverage_sh.plugin.CoverageParserThread") as mock_parser_thread, - mock.patch("coverage_sh.plugin.CoverageWriter"), - mock.patch("coverage_sh.plugin.MonitorThread"), - ): - config = CoverageConfig() - plugin.configure(config) - debug_output = debug.get_output() - # check DebugControl writes - assert "ShellPlugin.configure" in debug_output - assert mock_parser_thread.call_count == 1 - # check DebugControl is passed to CoverageParserThread - assert mock_parser_thread.call_args.kwargs["debug"] == debug From 6f093e79536eaad35ad0de231083506029ea2c51 Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Sat, 4 Apr 2026 15:48:14 +0200 Subject: [PATCH 13/15] fix and simplify tests --- coverage_sh/plugin.py | 8 ++++---- tests/test_plugin.py | 11 ++--------- 2 files changed, 6 insertions(+), 13 deletions(-) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 57fbdb4..5891482 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -66,14 +66,14 @@ def debug_write(msg: str, debug_control: DebugControl | None = None) -> None: # we are not recording coverage, so we have nowhere to send the message return - # DebugControl.write expects to be called from a frame with a "self" variable, so - # we use the same code to fetch that and pass it down to emulate that behavior - self = inspect.stack()[1][0].f_locals.get("self") # noqa: F841 - try: debug_control = debug_control or Coverage.current()._debug # type: ignore[union-attr] # noqa: SLF001 if debug_control.should(PLUGIN_DEBUG_OPTION): + # DebugControl.write expects to be called from a frame with a "self" variable, so + # we use the same code to fetch that and pass it down to emulate that behavior + self = inspect.stack()[1][0].f_locals.get("self") # noqa: F841 + debug_control.write(msg) except Exception as e: # noqa: BLE001 warn(f'Failed to log debug message: "{msg}": {e}', stacklevel=2) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 360b2c0..4686958 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -239,8 +239,6 @@ def test_end2end( class TestDebugWrite: def test_should_not_log_when_dsabled(self) -> None: debug_control = DebugControlString([]) - # we need a variable named `self` in the caller frame that is of a known type - self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 debug_write("foo", debug_control) @@ -248,8 +246,6 @@ def test_should_not_log_when_dsabled(self) -> None: def test_should_log_when_enabled(self) -> None: debug_control = DebugControlString(["shell"]) - # we need a variable named `self` in the caller frame that is of a known type - self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 debug_write("foo", debug_control) @@ -257,19 +253,16 @@ def test_should_log_when_enabled(self) -> None: def test_should_log_self_when_enabled(self) -> None: debug_control = DebugControlString(["self", "shell"]) - # we need a variable named `self` in the caller frame that is of a known type - self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 debug_write("foo", debug_control) assert ( - "self: None: debug_control = DebugControlString(["self", "callers", "shell"]) - # we need a variable named `self` in the caller frame that is of a known type - self = ShellPlugin({}) # type: ignore[assignment] # noqa: F841, PLW0642 debug_write("foo", debug_control) From 999a68dc9b9154ab243c9223aa9b31a0314451aa Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Sat, 4 Apr 2026 15:55:33 +0200 Subject: [PATCH 14/15] require only 90% coverage --- .github/workflows/lint_test.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/lint_test.yml b/.github/workflows/lint_test.yml index 63b830c..31a7560 100644 --- a/.github/workflows/lint_test.yml +++ b/.github/workflows/lint_test.yml @@ -66,7 +66,7 @@ jobs: pattern: coverage-data-* merge-multiple: true - - name: Combine coverage & fail if it's <100%. + - name: Combine coverage & fail if it's <90%. run: | python -Im pip install coverage[toml] @@ -78,8 +78,8 @@ jobs: # Report and write to summary. python -Im coverage report --format=markdown >> $GITHUB_STEP_SUMMARY - # Report again and fail if under 100%. - python -Im coverage report --fail-under=100 + # Report again and fail if under 90%. + python -Im coverage report --fail-under=90 export TOTAL=$(python -c "import json;print(json.load(open('coverage.json'))['totals']['percent_covered_display'])") echo "total=$TOTAL" >> $GITHUB_ENV From 3c2b97a6a345bc48aa1347e131ccb86c2f7db71e Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Sat, 4 Apr 2026 15:59:38 +0200 Subject: [PATCH 15/15] cleanup --- coverage_sh/plugin.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 5891482..04465c2 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -293,10 +293,6 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] super().__init__(*args, **kwargs) return - # minimal init for __repr__ to work for COVERAGE_DEBUG=self - self.returncode = None - self.args = args - debug_write("__init__") # convert args into kwargs