Group annotations for (almost) contiguous lines

coverage_comment/annotations.py

@@ -1,24 +1,99 @@
 from __future__ import annotations
 
+import dataclasses
+import functools
+import itertools
+import pathlib
+from collections.abc import Iterable
+
 from coverage_comment import coverage as coverage_module
-from coverage_comment import github
 
-MISSING_LINES_GROUP_TITLE = "Annotations of lines with missing coverage"
+MAX_ANNOTATION_GAP = 3
 
 
-def create_pr_annotations(
-    annotation_type: str, diff_coverage: coverage_module.DiffCoverage
-):
-    github.send_workflow_command(
-        command="group", command_value=MISSING_LINES_GROUP_TITLE
-    )
+@dataclasses.dataclass(frozen=True)
+class Annotation:
+    file: pathlib.Path
+    line_start: int
+    line_end: int
 
-    for file_path, file_diff_coverage in diff_coverage.files.items():
-        for missing_line in file_diff_coverage.violation_lines:
-            github.create_missing_coverage_annotation(
-                annotation_type=annotation_type,
-                file=file_path,
-                line=missing_line,
-            )
 
-    github.send_workflow_command(command="endgroup", command_value="")
+def compute_contiguous_groups(
+    values: list[int], separators: set[int], joiners: set[int]
+) -> list[tuple[int, int]]:
+    """
+    Given a list of (sorted) values, a list of separators and a list of
+    joiners, return a list of ranges (start, included end) describing groups of
+    values.
+
+    Groups are created by joining contiguous values together, and in some cases
+    by merging groups, enclosing a gap of values between them. Gaps that may be
+    enclosed are small gaps (<= MAX_ANNOTATION_GAP values after removing all
+    joiners) where no line is a "separator"
+    """
+    contiguous_groups: list[tuple[int, int]] = []
+    for _, contiguous_group in itertools.groupby(
+        zip(values, itertools.count(1)), lambda x: x[1] - x[0]
+    ):
+        grouped_values = (e[0] for e in contiguous_group)
+        first = next(grouped_values)
+        try:
+            *_, last = grouped_values
+        except ValueError:
+            last = first
+        contiguous_groups.append((first, last))
+
+    def reducer(
+        acc: list[tuple[int, int]], group: tuple[int, int]
+    ) -> list[tuple[int, int]]:
+        if not acc:
+            return [group]
+
+        last_group = acc[-1]
+        last_start, last_end = last_group
+        next_start, next_end = group
+
+        gap = set(range(last_end + 1, next_start)) - joiners
+
+        gap_is_small = len(gap) <= MAX_ANNOTATION_GAP
+        gap_contains_separators = gap & separators
+
+        if gap_is_small and not gap_contains_separators:
+            acc[-1] = (last_start, next_end)
+            return acc
+
+        acc.append(group)
+        return acc
+
+    return functools.reduce(reducer, contiguous_groups, [])
+
+
+def group_annotations(
+    coverage: coverage_module.Coverage,
+    diff_coverage: coverage_module.DiffCoverage,
+) -> Iterable[Annotation]:
+    for path, diff_file in diff_coverage.files.items():
+        coverage_file = coverage.files[path]
+
+        # Lines that are covered or excluded should not be considered for
+        # filling a gap between violation groups.
+        # (so, lines that can appear in a gap are lines that are missing, or
+        # lines that do not contain code: blank lines or lines containing comments)
+        separators = {
+            *coverage_file.executed_lines,
+            *coverage_file.excluded_lines,
+        }
+        # Lines that are added should be considered for filling a gap, unless
+        # they are separators.
+        joiners = set(diff_file.added_lines) - separators
+
+        for start, end in compute_contiguous_groups(
+            values=diff_file.missing_lines,
+            separators=separators,
+            joiners=joiners,
+        ):
+            yield Annotation(
+                file=path,
+                line_start=start,
+                line_end=end,
+            )

coverage_comment/coverage.py

@@ -5,7 +5,7 @@
 import decimal
 import json
 import pathlib
-from collections.abc import Iterable
+from collections.abc import Sequence
 
 from coverage_comment import log, subprocess
 
@@ -57,7 +57,13 @@ class Coverage:
 class FileDiffCoverage:
     path: pathlib.Path
     percent_covered: decimal.Decimal
-    violation_lines: list[int]
+    missing_lines: list[int]
+    added_lines: list[int]
+
+    # for backward compatibility
+    @property
+    def violation_lines(self) -> list[int]:
+        return self.missing_lines
 
 
 @dataclasses.dataclass
@@ -253,7 +259,8 @@ def get_diff_coverage_info(
         files[path] = FileDiffCoverage(
             path=path,
             percent_covered=percent_covered,
-            violation_lines=sorted(missing),
+            missing_lines=sorted(missing),
+            added_lines=added_lines_for_file,
         )
     final_percentage = compute_coverage(
         num_covered=total_num_lines - total_num_violations,
@@ -298,7 +305,7 @@ def parse_diff_output(diff: str) -> dict[pathlib.Path, list[int]]:
     return result
 
 
-def parse_line_number_diff_line(line: str) -> Iterable[int]:
+def parse_line_number_diff_line(line: str) -> Sequence[int]:
     """
     Parse the "added" part of the line number diff text:
         @@ -60,0 +61 @@ def compute_files(  -> [61]

coverage_comment/github.py

@@ -10,7 +10,6 @@
 from coverage_comment import github_client, log
 
 GITHUB_ACTIONS_LOGIN = "github-actions[bot]"
-MISSING_COVERAGE_MESSAGE = "This line has no coverage"
 
 
 class CannotDeterminePR(Exception):
@@ -199,17 +198,35 @@ def send_workflow_command(command: str, command_value: str, **kwargs: str) -> No
     )
 
 
-def create_missing_coverage_annotation(
-    annotation_type: str, file: pathlib.Path, line: int
+def create_missing_coverage_annotations(
+    annotation_type: str, annotations: list[tuple[pathlib.Path, int, int]]
 ):
+    """
+    Create annotations for lines with missing coverage.
+
+    annotation_type: The type of annotation to create. Can be either "error" or "warning".
+    annotations: A list of tuples of the form (file, line_start, line_end)
+    """
     send_workflow_command(
-        command=annotation_type,
-        command_value=MISSING_COVERAGE_MESSAGE,
-        # This will produce \ paths when running on windows.
-        # GHA doc is unclear whether this is right or not.
-        file=str(file),
-        line=str(line),
+        command="group", command_value="Annotations of lines with missing coverage"
     )
+    for file, line_start, line_end in annotations:
+        if line_start == line_end:
+            message = f"Missing coverage on line {line_start}"
+        else:
+            message = f"Missing coverage on lines {line_start}-{line_end}"
+
+        send_workflow_command(
+            command=annotation_type,
+            command_value=message,
+            # This will produce \ paths when running on windows.
+            # GHA doc is unclear whether this is right or not.
+            file=str(file),
+            line=str(line_start),
+            endLine=str(line_end),
+            title="Missing coverage",
+        )
+    send_workflow_command(command="endgroup", command_value="")
 
 
 def append_to_file(content: str, filepath: pathlib.Path):

coverage_comment/main.py

@@ -9,7 +9,9 @@
 
 from coverage_comment import activity as activity_module
 from coverage_comment import (
-    annotations,
+    annotations as annotations_module,
+)
+from coverage_comment import (
     comment_file,
     communication,
     files,
@@ -201,8 +203,15 @@ def process_pr(
             pr_number = None
 
     if pr_number is not None and config.ANNOTATE_MISSING_LINES:
-        annotations.create_pr_annotations(
-            annotation_type=config.ANNOTATION_TYPE, diff_coverage=diff_coverage
+        annotations = annotations_module.group_annotations(
+            coverage=coverage, diff_coverage=diff_coverage
+        )
+        github.create_missing_coverage_annotations(
+            annotation_type=config.ANNOTATION_TYPE,
+            annotations=[
+                (annotation.file, annotation.line_start, annotation.line_end)
+                for annotation in annotations
+            ],
         )
 
     try:

coverage_comment/settings.py

@@ -130,7 +130,7 @@ def GITHUB_PR_NUMBER(self) -> int | None:
 
     @property
     def GITHUB_BRANCH_NAME(self) -> str | None:
-        # "refs/head/my_branch_name"
+        # "refs/heads/my_branch_name"
         if "heads" in self.GITHUB_REF:
             return self.GITHUB_REF.split("/", 2)[2]
         return None

tests/conftest.py

@@ -245,7 +245,8 @@ def diff_coverage_obj():
             pathlib.Path("codebase/code.py"): coverage_module.FileDiffCoverage(
                 path=pathlib.Path("codebase/code.py"),
                 percent_covered=decimal.Decimal("0.8"),
-                violation_lines=[7, 9],
+                missing_lines=[7, 9],
+                added_lines=[7, 8, 9],
             )
         },
     )
@@ -262,12 +263,14 @@ def diff_coverage_obj_many_missing_lines():
             pathlib.Path("codebase/code.py"): coverage_module.FileDiffCoverage(
                 path=pathlib.Path("codebase/code.py"),
                 percent_covered=decimal.Decimal("0.8"),
-                violation_lines=[7, 9],
+                missing_lines=[7, 9],
+                added_lines=[7, 8, 9],
             ),
             pathlib.Path("codebase/main.py"): coverage_module.FileDiffCoverage(
                 path=pathlib.Path("codebase/code.py"),
                 percent_covered=decimal.Decimal("0.8"),
-                violation_lines=[1, 2, 8, 17],
+                missing_lines=[1, 2, 8, 17],
+                added_lines=[1, 2, 3, 4, 5, 6, 7, 8, 17],
             ),
         },
     )

tests/integration/test_github.py

@@ -63,8 +63,8 @@ def test_download_artifact(gh, session, zip_bytes):
         github=gh,
         repository="foo/bar",
         artifact_name="foo",
-        run_id="123",
-        filename="foo.txt",
+        run_id=123,
+        filename=pathlib.Path("foo.txt"),
     )
 
     assert result == "bar"
@@ -83,8 +83,8 @@ def test_download_artifact__no_artifact(gh, session):
             github=gh,
             repository="foo/bar",
             artifact_name="foo",
-            run_id="123",
-            filename="foo.txt",
+            run_id=123,
+            filename=pathlib.Path("foo.txt"),
         )
 
 
@@ -104,8 +104,8 @@ def test_download_artifact__no_file(gh, session, zip_bytes):
             github=gh,
             repository="foo/bar",
             artifact_name="foo",
-            run_id="123",
-            filename="bar.txt",
+            run_id=123,
+            filename=pathlib.Path("bar.txt"),
         )
 
 
@@ -303,6 +303,10 @@ def test_set_output(output_file):
     assert output_file.read_text() == "foo=true\n"
 
 
+def test_set_output__empty():
+    assert github.set_output(github_output=None, foo=True) is None
+
+
 def test_get_workflow_command():
     output = github.get_workflow_command(
         command="foo", command_value="bar", file="main.py", line="1", title="someTitle"
@@ -325,27 +329,6 @@ def test_send_workflow_command(capsys):
     assert output.err.strip() == "::foo file=main.py,line=1,title=someTitle::bar"
 
 
-def test_create_missing_coverage_annotation(capsys):
-    github.create_missing_coverage_annotation(
-        annotation_type="warning", file=pathlib.Path("test.py"), line=42
-    )
-    output = capsys.readouterr()
-    assert (
-        output.err.strip()
-        == "::warning file=test.py,line=42::This line has no coverage"
-    )
-
-
-def test_create_missing_coverage_annotation__annotation_type(capsys):
-    github.create_missing_coverage_annotation(
-        annotation_type="error", file=pathlib.Path("test.py"), line=42
-    )
-    output = capsys.readouterr()
-    assert (
-        output.err.strip() == "::error file=test.py,line=42::This line has no coverage"
-    )
-
-
 def test_add_job_summary(summary_file):
     github.add_job_summary(
         content="[job summary part 1]\n", github_step_summary=summary_file
@@ -356,3 +339,20 @@ def test_add_job_summary(summary_file):
         content="[job summary part 2]", github_step_summary=summary_file
     )
     assert summary_file.read_text() == "[job summary part 1]\n[job summary part 2]"
+
+
+def test_annotations(capsys):
+    github.create_missing_coverage_annotations(
+        annotation_type="warning",
+        annotations=[
+            (pathlib.Path("codebase/code.py"), 1, 3),
+            (pathlib.Path("codebase/main.py"), 5, 5),
+        ],
+    )
+
+    expected = """::group::Annotations of lines with missing coverage
+::warning file=codebase/code.py,line=1,endLine=3,title=Missing coverage::Missing coverage on lines 1-3
+::warning file=codebase/main.py,line=5,endLine=5,title=Missing coverage::Missing coverage on line 5
+::endgroup::"""
+    output = capsys.readouterr()
+    assert output.err.strip() == expected

tests/integration/test_main.py

@@ -568,7 +568,7 @@ def test_action__pull_request__annotations(
         git=git,
     )
     expected = """::group::Annotations of lines with missing coverage
-::warning file=foo.py,line=12::This line has no coverage
+::warning file=foo.py,line=12,endLine=12,title=Missing coverage::Missing coverage on line 12
 ::endgroup::"""
     output = capsys.readouterr()