From 5758877f3e2768957c10ca05df25798aab219ecf Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Sun, 15 Mar 2026 11:59:15 +0100 Subject: [PATCH 1/4] arcs PoC --- coverage_sh/plugin.py | 53 +++++++++++++++++++++++++++++++++++++++---- tests/test_plugin.py | 45 ++++++++++++++++++++++++++++++++---- 2 files changed, 90 insertions(+), 8 deletions(-) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index 77d328b..c779f59 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -86,6 +86,7 @@ def __init__(self, filename: str) -> None: self.path = Path(filename) self._content: str | None = None self._executable_lines: set[int] = set() + self._arcs: set[tuple[int, int]] = set() self._translate_lines: dict[int, int] = {} self._parser = Parser(Language(tree_sitter_bash.language())) @@ -100,7 +101,12 @@ def source(self) -> str: return self._content - def _parse_ast(self, node: Node) -> None: + def _parse_ast( + self, + node: Node, + executable_parent: Node | None = None, + previous_executable: Node | None = None, + ) -> Node | None: if node.is_named and node.type in EXECUTABLE_NODE_TYPES: sline = node.start_point.row + 1 eline = node.end_point.row + 1 @@ -110,21 +116,60 @@ def _parse_ast(self, node: Node) -> None: for index in range(sline + 1, eline + 1): self._translate_lines[index] = sline - for child in node.children: - self._parse_ast(child) + if previous_executable is None: + # first executable node in the script + self._arcs.add((0, sline)) + else: + self._arcs.add((previous_executable.start_point.row + 1, sline)) - def lines(self) -> set[TLineNo]: + executable_parent = node + previous_executable = node + + for child in node.children: + if node.type == "function_definition": + # Function bodies are independent arc graphs: arcs inside the + # body must not connect to the call-site context, and the + # call-site context must not be affected by what happens inside. + self._parse_ast( + child, + executable_parent=node, + previous_executable=None, + ) + else: + previous_executable = self._parse_ast( + child, + executable_parent=executable_parent, + # Each direct child of an executable node starts fresh from + # that node, so alternative branches (e.g. else) arc from + # the parent rather than from the last sibling branch. + previous_executable=node + if node is executable_parent + else previous_executable, + ) + + return previous_executable + + def _ensure_parsed(self) -> None: + if self._executable_lines: + return # already parsed tree = self._parser.parse(self.source().encode("utf-8")) self._parse_ast(tree.root_node) + def lines(self) -> set[TLineNo]: + self._ensure_parsed() return self._executable_lines def translate_lines(self, input_lines: Iterable[TLineNo]) -> set[TLineNo]: + self._ensure_parsed() result: set[TLineNo] = set() for index in input_lines: result.add(self._translate_lines.get(index, index)) return result + def arcs(self) -> set[tuple[TLineNo, TLineNo]]: + self._ensure_parsed() + return {(src, dst) for src, dst in self._arcs if src != dst} + def filename_suffix() -> str: die = Random(os.urandom(8)) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index a6afb2d..c53df4a 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -100,9 +100,7 @@ def test_end2end( monkeypatch: pytest.MonkeyPatch, cover_always: bool, ) -> None: - test_sh = tmp_path / "test.sh" - test_sh.write_text("#!/bin/bash\necho hello\n") - test_sh.chmod(0o755) + test_sh = Path(__file__).parent / "resources" / "syntax_example.sh" pyproject_toml = tmp_path / "pyproject.toml" pyproject_toml.write_text( @@ -110,7 +108,7 @@ def test_end2end( ) main_py = tmp_path / "main.py" - main_py.write_text("import subprocess\nsubprocess.run(['./test.sh'])\n") + main_py.write_text(f"import subprocess\nsubprocess.run(['{test_sh}'])\n") if cover_always: with pyproject_toml.open("a") as fd: @@ -258,6 +256,45 @@ def test_executable_lines( reporter = ShellFileReporter(str(script)) assert reporter.lines() == expected_lines + @pytest.mark.parametrize( + ("script_body", "expected_arcs"), + [ + pytest.param( + "echo one\necho two\necho three\n", + {(0, 2), (2, 3), (3, 4)}, + id="sequential", + ), + pytest.param( + "if true; then\n echo yes\nelse\n echo no\nfi\n", + {(0, 2), (2, 3), (2, 5)}, + id="if_else", + ), + pytest.param( + "if true; then\n echo yes\nfi\necho after\n", + {(0, 2), (2, 3), (2, 5)}, + id="if_no_else", + ), + pytest.param( + "for i in 1 2; do\n echo $i\ndone\necho after\n", + {(0, 2), (2, 3), (3, 5)}, + id="for_loop", + ), + pytest.param( + "echo before\nfunction say_hello() {\n echo hello\n echo world\n}\nsay_hello\n", + {(0, 2), (0, 4), (2, 7), (4, 5)}, + id="function_definition", + ), + ], + ) + def test_arcs( + self, tmp_path: Path, script_body: str, expected_arcs: set[tuple[int, int]] + ) -> None: + # Each script has a shebang on line 1; the construct under test starts on line 2. + script = tmp_path / "script.sh" + script.write_text(f"#!/bin/bash\n{script_body}") + reporter = ShellFileReporter(str(script)) + assert reporter.arcs() == expected_arcs + def test_invalid_syntax_should_be_treated_as_executable( self, tmp_path: Path ) -> None: From 722d804157fd3aa73cb07a4f487618712e030159 Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Sat, 21 Mar 2026 09:41:14 +0100 Subject: [PATCH 2/4] add brnach coverage exampl --- example/syntax_example.sh | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/example/syntax_example.sh b/example/syntax_example.sh index 3791da3..080ff31 100755 --- a/example/syntax_example.sh +++ b/example/syntax_example.sh @@ -64,6 +64,14 @@ case $fruit in ;; esac +# branch coverage example +selected="" +if [[ $fruit == "banana" ]]; then + selected="banana" +fi +echo $selected + + echo multi \ line echo \ From 77502bf0cb03f819e732972392e97012e8e47669 Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Sat, 21 Mar 2026 09:53:01 +0100 Subject: [PATCH 3/4] ShellFileReporter: finished arcs --- coverage_sh/plugin.py | 145 ++++++++++++++++++++++++++++++++++++++++-- tests/test_plugin.py | 57 ++++++++++++++--- 2 files changed, 188 insertions(+), 14 deletions(-) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index c779f59..f5c0dff 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -55,6 +55,7 @@ "list", } SUPPORTED_MIME_TYPES = {"text/x-shellscript"} +MAX_BRANCH_EXITS = 2 PLUGIN_DEBUG_OPTION = "shell" @@ -87,9 +88,51 @@ def __init__(self, filename: str) -> None: self._content: str | None = None self._executable_lines: set[int] = set() self._arcs: set[tuple[int, int]] = set() + self._no_branch_lines: set[int] = set() self._translate_lines: dict[int, int] = {} self._parser = Parser(Language(tree_sitter_bash.language())) + def _is_exhaustive_if_statement(self, node: Node) -> bool: + """Return true when an if statement has an else branch. + + We use this to detect control-flow nodes where all outcomes are handled + inside the block, so a direct fallthrough arc from the if header to the + next statement would be incorrect. + """ + if node.type != "if_statement": + return False + return any(child.type == "else_clause" for child in node.children) + + def _is_exhaustive_case_statement(self, node: Node) -> bool: + """Return true when a case statement has a default ``*)`` pattern. + + Tree-sitter exposes case patterns as ``case_item`` children. A default + arm is represented by an ``extglob_pattern`` node with text ``*``. + """ + if node.type != "case_statement": + return False + + for child in node.children: + if child.type != "case_item": + continue + + first_named_child = next( + ( + case_item_child + for case_item_child in child.children + if case_item_child.is_named + ), + None, + ) + if ( + first_named_child is not None + and first_named_child.type == "extglob_pattern" + and first_named_child.text == b"*" + ): + return True + + return False + def source(self) -> str: if self._content is None: if not self.path.is_file(): @@ -107,45 +150,74 @@ def _parse_ast( executable_parent: Node | None = None, previous_executable: Node | None = None, ) -> Node | None: + # For exhaustive control nodes (if+else / case+*), we keep track of the + # last executable statement reached in any branch so callers can chain + # control-flow from inside the block, not from the header line. + exhaustive_control = False + branch_last_executable: Node | None = previous_executable + if node.is_named and node.type in EXECUTABLE_NODE_TYPES: sline = node.start_point.row + 1 eline = node.end_point.row + 1 self._executable_lines.add(sline) + + exhaustive_control = self._is_exhaustive_if_statement( + node + ) or self._is_exhaustive_case_statement(node) + # for multi-line commands translate to the first line if sline != eline and node.type == "command": for index in range(sline + 1, eline + 1): self._translate_lines[index] = sline if previous_executable is None: - # first executable node in the script - self._arcs.add((0, sline)) + # first executable node in file / function + self._arcs.add((-1, sline)) else: self._arcs.add((previous_executable.start_point.row + 1, sline)) executable_parent = node previous_executable = node + branch_last_executable = node for child in node.children: if node.type == "function_definition": # Function bodies are independent arc graphs: arcs inside the # body must not connect to the call-site context, and the # call-site context must not be affected by what happens inside. - self._parse_ast( + func_last = self._parse_ast( child, executable_parent=node, previous_executable=None, ) + if func_last is not None: + self._arcs.add((func_last.start_point.row + 1, -1)) else: - previous_executable = self._parse_ast( + child_last = self._parse_ast( child, executable_parent=executable_parent, # Each direct child of an executable node starts fresh from # that node, so alternative branches (e.g. else) arc from # the parent rather than from the last sibling branch. + # This avoids creating fake sequential arcs between sibling + # branches that are mutually exclusive. previous_executable=node if node is executable_parent else previous_executable, ) + previous_executable = child_last + if ( + node is executable_parent + and child_last is not None + and child_last is not node + ): + branch_last_executable = child_last + + if node is executable_parent and exhaustive_control: + # For exhaustive controls, returning the branch tail prevents a + # synthetic fallthrough arc from the control header to the next + # statement after the block. + return branch_last_executable return previous_executable @@ -154,15 +226,68 @@ def _ensure_parsed(self) -> None: return # already parsed tree = self._parser.parse(self.source().encode("utf-8")) self._parse_ast(tree.root_node) + self._collapse_multiway_exits() + if self._executable_lines: + self._arcs.add((max(self._executable_lines), -1)) + + def _collapse_multiway_exits(self) -> None: + """Collapse multi-way branch exits to binary form for HTML compatibility. + + Coverage.py's HTML reporter has an assertion that each branch line has at + most one "long" annotation (the verbose description of a missing arc). This + assertion fails for shell scripts with multi-way branches like: + + case $var in + a) echo A ;; + b) echo B ;; + c) echo C ;; + *) echo D ;; + esac + + The AST parser produces a single branch line (the "case" keyword) with 4 + exits to each case arm. When coverage reports missing branches, this creates + 3 long annotations -> assertion failure. + + Instead of modeling the case statement as one line with N exits, we collapse + it to a binary form: keep only the first and last exit destinations, and mark + the source line as "no branch" so coverage won't emit branch annotations. + + Implementation: + - Scan all arcs to find source lines with > MAX_BRANCH_EXITS (2) destinations + - For each such line, keep only the min and max destination arcs + - Record the source line in _no_branch_lines so exit_counts still reports 2 + but HTML won't annotate it as a multi-way branch + """ + exits_by_line: dict[int, set[int]] = defaultdict(set) + for src, dst in self._arcs: + if src > 0 and dst > 0 and dst != src: + exits_by_line[src].add(dst) + + for src, exits in exits_by_line.items(): + if len(exits) <= MAX_BRANCH_EXITS: + continue + + self._no_branch_lines.add(src) + sorted_exits = sorted(exits) + keep = {sorted_exits[0], sorted_exits[-1]} + self._arcs = { + (arc_src, arc_dst) + for arc_src, arc_dst in self._arcs + if arc_src != src or arc_dst <= 0 or arc_dst in keep + } + + def no_branch_lines(self) -> set[TLineNo]: + self._ensure_parsed() + return self._no_branch_lines def lines(self) -> set[TLineNo]: self._ensure_parsed() return self._executable_lines - def translate_lines(self, input_lines: Iterable[TLineNo]) -> set[TLineNo]: + def translate_lines(self, lines: Iterable[TLineNo]) -> set[TLineNo]: self._ensure_parsed() result: set[TLineNo] = set() - for index in input_lines: + for index in lines: result.add(self._translate_lines.get(index, index)) return result @@ -170,6 +295,14 @@ def arcs(self) -> set[tuple[TLineNo, TLineNo]]: self._ensure_parsed() return {(src, dst) for src, dst in self._arcs if src != dst} + def exit_counts(self) -> dict[TLineNo, int]: + self._ensure_parsed() + exits: dict[TLineNo, set[TLineNo]] = defaultdict(set) + for src, dst in self.arcs(): + if src > 0: + exits[src].add(dst) + return {src: len(dsts) for src, dsts in exits.items() if len(dsts) > 1} + def filename_suffix() -> str: die = Random(os.urandom(8)) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index c53df4a..2797b1f 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,5 +1,7 @@ # SPDX-License-Identifier: MIT # Copyright (c) 2023-2024 Kilian Lackhove +from __future__ import annotations + import asyncio import io import os @@ -9,12 +11,11 @@ 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 from time import sleep -from typing import cast +from typing import TYPE_CHECKING, cast import coverage import pytest @@ -35,6 +36,9 @@ filename_suffix, ) +if TYPE_CHECKING: + from collections.abc import Iterable + COVERAGE_LINE_CHUNKS = ( b"""\ CCOV:::/home/dummy_user/dummy_dir_a:::1:::a normal line @@ -100,7 +104,7 @@ def test_end2end( monkeypatch: pytest.MonkeyPatch, cover_always: bool, ) -> None: - test_sh = Path(__file__).parent / "resources" / "syntax_example.sh" + test_sh = Path(__file__).parent.parent / "example" / "syntax_example.sh" pyproject_toml = tmp_path / "pyproject.toml" pyproject_toml.write_text( @@ -261,27 +265,32 @@ def test_executable_lines( [ pytest.param( "echo one\necho two\necho three\n", - {(0, 2), (2, 3), (3, 4)}, + {(-1, 2), (2, 3), (3, 4), (4, -1)}, id="sequential", ), pytest.param( "if true; then\n echo yes\nelse\n echo no\nfi\n", - {(0, 2), (2, 3), (2, 5)}, + {(-1, 2), (2, 3), (2, 5), (5, -1)}, id="if_else", ), + pytest.param( + "if true; then\n echo yes\nelse\n echo no\nfi\necho after\n", + {(-1, 2), (2, 3), (2, 5), (5, 7), (7, -1)}, + id="if_else_with_following_statement", + ), pytest.param( "if true; then\n echo yes\nfi\necho after\n", - {(0, 2), (2, 3), (2, 5)}, + {(-1, 2), (2, 3), (2, 5), (5, -1)}, id="if_no_else", ), pytest.param( "for i in 1 2; do\n echo $i\ndone\necho after\n", - {(0, 2), (2, 3), (3, 5)}, + {(-1, 2), (2, 3), (3, 5), (5, -1)}, id="for_loop", ), pytest.param( "echo before\nfunction say_hello() {\n echo hello\n echo world\n}\nsay_hello\n", - {(0, 2), (0, 4), (2, 7), (4, 5)}, + {(-1, 2), (-1, 4), (2, 7), (4, 5), (5, -1), (7, -1)}, id="function_definition", ), ], @@ -295,6 +304,38 @@ def test_arcs( reporter = ShellFileReporter(str(script)) assert reporter.arcs() == expected_arcs + def test_exit_counts_should_collapse_multiway_if_elif_branches( + self, tmp_path: Path + ) -> None: + script = tmp_path / "script.sh" + script.write_text( + "#!/bin/bash\n" + "if false; then\n" + " echo if\n" + "elif false; then\n" + " echo elif\n" + "else\n" + " echo else\n" + "fi\n" + ) + reporter = ShellFileReporter(str(script)) + assert reporter.exit_counts() == {2: 2} + assert reporter.no_branch_lines() == {2} + + def test_exit_counts_should_include_case_branches(self, tmp_path: Path) -> None: + script = tmp_path / "script.sh" + script.write_text( + "#!/bin/bash\n" + "case $x in\n" + " a) echo A ;;\n" + " b) echo B ;;\n" + " *) echo D ;;\n" + "esac\n" + ) + reporter = ShellFileReporter(str(script)) + assert reporter.exit_counts() == {2: 2} + assert reporter.no_branch_lines() == {2} + def test_invalid_syntax_should_be_treated_as_executable( self, tmp_path: Path ) -> None: From 7c4febb9feb58fcfbd2c35612ae890cebbd1d861 Mon Sep 17 00:00:00 2001 From: Kilian Lackhove Date: Tue, 31 Mar 2026 21:48:28 +0200 Subject: [PATCH 4/4] CovLineParser: finished arcs --- coverage_sh/plugin.py | 92 ++++++++++++++++++++++++++++++++++++------ tests/test_plugin.py | 94 +++++++++++++++++++++++++++++++++++++------ 2 files changed, 162 insertions(+), 24 deletions(-) diff --git a/coverage_sh/plugin.py b/coverage_sh/plugin.py index f5c0dff..95e1b07 100644 --- a/coverage_sh/plugin.py +++ b/coverage_sh/plugin.py @@ -33,6 +33,7 @@ from tree_sitter import Node LineData = dict[str, set[int]] +ArcData = dict[str, set[tuple[int, int]]] TMP_PATH = Path(os.environ.get("XDG_RUNTIME_DIR", "/tmp")) # noqa: S108 TRACEFILE_PREFIX = "shelltrace" @@ -314,7 +315,11 @@ def filename_suffix() -> str: class CovLineParser: def __init__(self) -> None: self._last_line_fragment = "" + self._last_line = -1 + self._last_path = "" + self._last_function = "" self.line_data: LineData = defaultdict(set) + self.arc_data: ArcData = defaultdict(set) def parse(self, buf: bytes) -> None: self._report_lines(list(self._buf_to_lines(buf))) @@ -340,39 +345,75 @@ def _report_lines(self, lines: list[str]) -> None: continue try: - _, path_, lineno_, _ = line.split(":::", maxsplit=3) + path_, lineno_, func_ = self._parse_trace_line(line) lineno = int(lineno_) path = Path(path_).absolute() except ValueError as e: raise ValueError(f"could not parse line {line}") from e - self.line_data[str(path)].add(lineno) + path_str = str(path) + self.line_data[path_str].add(lineno) + + if self._last_path == "": + # first line ever + self.arc_data[path_str].add((-1, lineno)) + elif path_str == self._last_path: + # same file + if func_ != self._last_function: + # function scope changed + self.arc_data[self._last_path].add((self._last_line, -1)) + self.arc_data[path_str].add((-1, lineno)) + else: + # same function + self.arc_data[path_str].add((self._last_line, lineno)) + else: + # different file + self.arc_data[self._last_path].add((self._last_line, -1)) + self.arc_data[path_str].add((-1, lineno)) + + self._last_line = lineno + self._last_path = path_str + self._last_function = func_ + + def _parse_trace_line(self, line: str) -> tuple[str, str, str]: + _, path_, lineno_, func_ = line.split(":::", maxsplit=3) + func_ = func_.split(":::", 1)[0] + return (path_, lineno_, func_) def flush(self) -> None: self.parse(b"\n") + def finalize(self) -> None: + self.arc_data[self._last_path].add((self._last_line, -1)) + class CoverageWriter: - def __init__(self, coverage_data_path: Path): + def __init__(self, coverage_data_path: Path, *, branch: bool = False): # pytest-cov uses the COV_CORE_DATAFILE env var to configure the datafile base path coverage_data_env_var = os.environ.get("COV_CORE_DATAFILE") if coverage_data_env_var is not None: coverage_data_path = Path(coverage_data_env_var).absolute() self._coverage_data_path = coverage_data_path + self._branch = branch - def write(self, line_data: LineData) -> None: + def write(self, line_data: LineData, arc_data: ArcData | None = None) -> None: suffix_ = "sh." + filename_suffix() coverage_data = CoverageData( basename=self._coverage_data_path, suffix=suffix_, - # TODO: set warn, debug and no_disk ) coverage_data.add_file_tracers( dict.fromkeys(line_data, "coverage_sh.ShellPlugin") ) - coverage_data.add_lines(line_data) + if self._branch: + if arc_data: + for path, arcs in arc_data.items(): + if arcs: + coverage_data.add_arcs({path: arcs}) + else: + coverage_data.add_lines(line_data) coverage_data.write() @@ -437,7 +478,11 @@ def run(self) -> None: sel.unregister(fifo) os.close(fifo) - self._coverage_writer.write(self._parser.line_data) + self._parser.finalize() + + self._coverage_writer.write( + self._parser.line_data, arc_data=self._parser.arc_data + ) with contextlib.suppress(FileNotFoundError): self.fifo_path.unlink() @@ -449,7 +494,7 @@ def init_helper(fifo_path: Path) -> Path: helper_path = Path(TMP_PATH, f"coverage-sh.{filename_suffix()}.sh") helper_path.write_text( rf"""#!/bin/sh -PS4="COV:::\${{BASH_SOURCE}}:::\${{LINENO}}:::" +PS4="COV:::\${{BASH_SOURCE}}:::\${{LINENO}}:::\${{FUNCNAME[0]}}:::" exec {{BASH_XTRACEFD}}>>"{fifo_path!s}" export BASH_XTRACEFD set -x @@ -463,6 +508,7 @@ def init_helper(fifo_path: Path) -> Path: # ignore this for the time being class PatchedPopen(OriginalPopen): # type: ignore[type-arg] data_file_path: Path = Path.cwd() + branch: bool = False def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] if Coverage.current() is None: @@ -478,7 +524,9 @@ def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] kwargs.update(dict(zip(sig.parameters.keys(), args))) self._parser_thread = CoverageParserThread( - coverage_writer=CoverageWriter(coverage_data_path=self.data_file_path), + coverage_writer=CoverageWriter( + coverage_data_path=self.data_file_path, branch=self.branch + ), name="CoverageParserThread(None)", ) self._parser_thread.start() @@ -543,6 +591,21 @@ def _iterdir(path: Path) -> Iterator[Path]: yield from _iterdir(p) +class ShellFileTracer(FileTracer): + def __init__(self, filename: str) -> None: + super().__init__(filename) # type: ignore[call-arg] + self._reporter = ShellFileReporter(filename) + + def source(self) -> str: + return self._reporter.source() + + def source_token_lines(self) -> Iterable[object]: + return [] # pragma: no cover + + def find_executable_statements(self, _source: str, _filename: str) -> set[int]: + return self._reporter.lines() + + class ShellPlugin(CoveragePlugin): def __init__(self, options: dict[str, Any]): self.options = options @@ -551,6 +614,7 @@ def __init__(self, options: dict[str, Any]): 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() + branch = bool(config.get_option("run:branch")) if config.get_option("run:core") == "sysmon" or ( sys.version_info >= (3, 14) and config.get_option("run:core") is None @@ -564,7 +628,7 @@ def configure(self, config: TConfigurable) -> None: if self.options.get("cover_always", False): parser_thread = CoverageParserThread( - coverage_writer=CoverageWriter(coverage_data_path), + coverage_writer=CoverageWriter(coverage_data_path, branch=branch), name=f"CoverageParserThread({coverage_data_path!s})", ) parser_thread.start() @@ -579,6 +643,7 @@ def configure(self, config: TConfigurable) -> None: os.environ["ENV"] = str(self._helper_path) else: PatchedPopen.data_file_path = coverage_data_path + PatchedPopen.branch = branch # https://github.com/python/mypy/issues/1152 subprocess.Popen = PatchedPopen # type: ignore[misc] @@ -591,8 +656,11 @@ def __del__(self) -> None: def _is_relevant(path: Path) -> bool: return magic.from_file(path.resolve(), mime=True) in SUPPORTED_MIME_TYPES - def file_tracer(self, filename: str) -> FileTracer | None: # noqa: ARG002 - return None + def file_tracer(self, filename: str) -> FileTracer | None: + path = Path(filename) + if not path.exists() or not self._is_relevant(path): + return None + return ShellFileTracer(filename) def file_reporter( self, diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 2797b1f..2bae771 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -24,6 +24,7 @@ from packaging.version import Version from coverage_sh.plugin import ( + ArcData, CoverageParserThread, CoverageWriter, CovLineParser, @@ -41,12 +42,12 @@ COVERAGE_LINE_CHUNKS = ( b"""\ -CCOV:::/home/dummy_user/dummy_dir_a:::1:::a normal line -COV:::/home/dummy_user/dummy_dir_b:::10:::a line +CCOV:::/home/dummy_user/dummy_dir_a:::1:::main:::a normal line +COV:::/home/dummy_user/dummy_dir_b:::10:::main:::a line with a line fragment -COV:::/home/dummy_user/dummy_dir_a:::2:::a line with ::: triple columns -COV:::/home/dummy_user/dummy_dir_a:::3:::a line """, +COV:::/home/dummy_user/dummy_dir_a:::2:::main:::a line with ::: triple columns +COV:::/home/dummy_user/dummy_dir_a:::3:::main:::a line """, b"that spans multiple chunks\n", b"C", b"O", @@ -58,20 +59,36 @@ b"ho", b"m", b"e", - b"/dummy_user/dummy_dir_a:::4:::a chunked line", + b"/dummy_user/dummy_dir_a:::18:::some_func:::a chunked line\n", + b"COV:::/home/dummy_user/dummy_dir_a:::4:::main:::final line", ) COVERAGE_LINES = [ - "CCOV:::/home/dummy_user/dummy_dir_a:::1:::a normal line", - "COV:::/home/dummy_user/dummy_dir_b:::10:::a line", + "CCOV:::/home/dummy_user/dummy_dir_a:::1:::main:::a normal line", + "COV:::/home/dummy_user/dummy_dir_b:::10:::main:::a line", "with a line fragment", - "COV:::/home/dummy_user/dummy_dir_a:::2:::a line with ::: triple columns", - "COV:::/home/dummy_user/dummy_dir_a:::3:::a line that spans multiple chunks", - "COV:::/home/dummy_user/dummy_dir_a:::4:::a chunked line", + "COV:::/home/dummy_user/dummy_dir_a:::2:::main:::a line with ::: triple columns", + "COV:::/home/dummy_user/dummy_dir_a:::3:::main:::a line that spans multiple chunks", + "COV:::/home/dummy_user/dummy_dir_a:::18:::some_func:::a chunked line", + "COV:::/home/dummy_user/dummy_dir_a:::4:::main:::final line", ] COVERAGE_LINE_COVERAGE = { - "/home/dummy_user/dummy_dir_a": {1, 2, 3, 4}, + "/home/dummy_user/dummy_dir_a": {1, 2, 3, 4, 18}, "/home/dummy_user/dummy_dir_b": {10}, } +COVERAGE_ARC_COVERAGE = { + "/home/dummy_user/dummy_dir_a": { + (-1, 1), + (1, -1), + (-1, 2), + (2, 3), + (3, -1), + (-1, 18), + (18, -1), + (-1, 4), + (4, -1), + }, + "/home/dummy_user/dummy_dir_b": {(-1, 10), (10, -1)}, +} END2END_SUBPROCESS_TIMEOUT = 5 @@ -402,6 +419,54 @@ def test_parse_result_matches_reference(self) -> None: assert parser.line_data == COVERAGE_LINE_COVERAGE + @pytest.mark.parametrize( + ("chunks", "expected_arcs"), + [ + pytest.param( + COVERAGE_LINE_CHUNKS, + COVERAGE_ARC_COVERAGE, + id="chunked_lines", + ), + pytest.param( + (b"COV:::/path/a:::1:::x\nCOV:::/path/a:::2:::x\n",), + {"/path/a": {(-1, 1), (1, 2), (2, -1)}}, + id="single_file_sequential", + ), + pytest.param( + ( + b"COV:::/path/a:::5:::x\nCOV:::/path/a:::3:::x\nCOV:::/path/a:::7:::x\n", + ), + {"/path/a": {(-1, 5), (5, 3), (3, 7), (7, -1)}}, + id="single_file_non_sequential", + ), + pytest.param( + ( + b"COV:::/path/a:::5:::x\nCOV:::/path/b:::3:::x\nCOV:::/path/b:::7:::x\n", + ), + { + "/path/a": {(-1, 5), (5, -1)}, + "/path/b": {(-1, 3), (3, 7), (7, -1)}, + }, + id="multi_file", + ), + pytest.param( + (b"COV:::/path/single:::1:::x\n",), + {"/path/single": {(-1, 1), (1, -1)}}, + id="single_line_no_arcs", + ), + ], + ) + def test_arcs_should_match_expected( + self, chunks: tuple[bytes, ...], expected_arcs: ArcData + ) -> None: + parser = CovLineParser() + for chunk in chunks: + parser.parse(chunk) + parser.flush() + parser.finalize() + + assert parser.arc_data == expected_arcs + def test_parse_should_raise_for_incomplete_line(self) -> None: parser = CovLineParser() with pytest.raises(ValueError, match="could not parse line"): @@ -441,7 +506,12 @@ class CovWriterFake: def __init__(self) -> None: self.line_data: LineData = defaultdict(set) - def write(self, line_data: LineData) -> None: + def write( + self, + line_data: LineData, + *, + arc_data: ArcData | None = None, # noqa: ARG002 + ) -> None: self.line_data.update(line_data) def test_lines_should_match_reference(self) -> None: