fix: multiline statement branches with sysmon #2070

Some cases of conditions or for loops were reported as missing branches
when using sys.monitoring, which is now the default in 3.14.

Author: nedbat

CHANGES.rst

@@ -23,10 +23,16 @@ upgrading your version of coverage.py.
 Unreleased
 ----------
 
+- Fix: some multi-line case clauses or for loops (and probably other
+  constructs) could cause incorrect claims of missing branches with the
+  sys.monitoring core, as described in `issue 2070`_. This is now fixed.
+
 - Split ``sqlite`` debugging information out of the ``sys`` :ref:`coverage
   debug <cmd_debug>` and :ref:`cmd_run_debug` options since it's bulky and not
   very useful.
 
+.. _issue 2070: https://github.com/nedbat/coveragepy/issues/2070
+
 
 .. start-releases
 

coverage/bytecode.py

@@ -10,7 +10,7 @@
 from types import CodeType
 from typing import Iterable, Optional
 
-from coverage.types import TArc, TOffset
+from coverage.types import TArc, TLineNo, TOffset
 
 
 def code_objects(code: CodeType) -> Iterable[CodeType]:
@@ -99,7 +99,7 @@ def walk(
 TBranchTrails = dict[TOffset, TBranchTrailsOneSource]
 
 
-def branch_trails(code: CodeType) -> TBranchTrails:
+def branch_trails(code: CodeType, multiline_map: dict[TLineNo, TLineNo]) -> TBranchTrails:
     """
     Calculate branch trails for `code`.
 
@@ -128,6 +128,7 @@ def branch_trails(code: CodeType) -> TBranchTrails:
         from_line = inst.line_number
         if from_line is None:
             continue
+        from_line = multiline_map.get(from_line, from_line)
 
         def add_one_branch_trail(
             trails: TBranchTrailsOneSource,
@@ -138,8 +139,11 @@ def add_one_branch_trail(
             to_line = None
             for inst2 in iwalker.walk(start_at=start_at, follow_jumps=True):
                 inst_offsets.add(inst2.offset)
-                if inst2.line_number and inst2.line_number != from_line:
-                    to_line = inst2.line_number
+                l2 = inst2.line_number
+                if l2 is not None:
+                    l2 = multiline_map.get(l2, l2)
+                if l2 and l2 != from_line:
+                    to_line = l2
                     break
                 elif inst2.jump_target and (inst2.opcode not in ALWAYS_JUMPS):
                     break

coverage/sysmon.py

@@ -20,6 +20,7 @@
 from coverage.bytecode import TBranchTrails, always_jumps, branch_trails
 from coverage.debug import short_filename, short_stack
 from coverage.misc import isolate_module
+from coverage.parser import PythonParser
 from coverage.types import (
     AnyCallable,
     TFileDisposition,
@@ -428,7 +429,8 @@ def sysmon_branch_either(
         if not code_info.branch_trails:
             if self.stats is not None:
                 self.stats["branch_trails"] += 1
-            code_info.branch_trails = branch_trails(code)
+            multiline_map = get_multiline_map(code.co_filename)
+            code_info.branch_trails = branch_trails(code, multiline_map=multiline_map)
             code_info.always_jumps = always_jumps(code)
             # log(f"branch_trails for {code}:\n{ppformat(code_info.branch_trails)}")
         added_arc = False
@@ -462,3 +464,11 @@ def sysmon_branch_either(
                 # log(f"adding unforeseen {arc=}")
 
         return DISABLE
+
+
+@functools.lru_cache(maxsize=5)
+def get_multiline_map(filename: str) -> dict[TLineNo, TLineNo]:
+    """Get a PythonParser for the given filename, cached."""
+    parser = PythonParser(filename=filename)
+    parser.parse_source()
+    return parser.multiline_map

tests/test_arcs.py

@@ -720,6 +720,19 @@ def forelse(seq):
             branchz_missing="26",
         )
 
+    def test_split_for(self) -> None:
+        self.check_coverage(
+            """\
+            a = 0
+            for (i
+                ) in [1,2,3,4,5]:
+                a += i
+            assert a == 15
+            """,
+            lines=[1, 2, 4, 5],
+            branchz="24 25",
+        )
+
     def test_while_else(self) -> None:
         self.check_coverage(
             """\
@@ -1637,6 +1650,30 @@ def absurd(x):
         )
         assert self.stdout() == "also not default\n"
 
+    def test_split_match_case(self) -> None:
+        self.check_coverage(
+            """\
+            def foo(x):
+                match x:
+                    case (
+                        1
+                        | 2
+                    ):
+                        return "output: 1 or 2"
+                    case _:
+                        return "output: other"
+
+            print(foo(1))
+            print(foo(2))
+            print(foo(3))
+            """,
+            lines=[1, 2, 3, 7, 8, 9, 11, 12, 13],
+            missing="",
+            branchz="37 38",
+            branchz_missing="",
+        )
+        assert self.stdout() == "output: 1 or 2\noutput: 1 or 2\noutput: other\n"
+
 
 class OptimizedIfTest(CoverageTest):
     """Tests of if statements being optimized away."""