- initial port to Python 3.12, using PEP669. Pytest interposition isn't

  supported (yet), and line numbers in branches are limited in range;

setup.py

@@ -100,7 +100,7 @@ def limited_api_args():
     packages=['slipcover'],
     package_dir={'': 'src'},
     ext_modules=([probe]),
-    python_requires=">=3.8,<3.12",
+    python_requires=">=3.8,<3.13",
     install_requires=[
         "tabulate"
     ],

src/slipcover/__main__.py

@@ -36,7 +36,8 @@
 ap.add_argument('--silent', action='store_true', help=argparse.SUPPRESS)
 ap.add_argument('--dis', action='store_true', help=argparse.SUPPRESS)
 ap.add_argument('--debug', action='store_true', help=argparse.SUPPRESS)
-ap.add_argument('--dont-wrap-pytest', action='store_true', help=argparse.SUPPRESS)
+if sys.version_info[0:2] < (3,12):
+    ap.add_argument('--dont-wrap-pytest', action='store_true', help=argparse.SUPPRESS)
 
 g = ap.add_mutually_exclusive_group(required=True)
 g.add_argument('-m', dest='module', nargs=1, help="run given module as __main__")
@@ -72,7 +73,7 @@
                    skip_covered=args.skip_covered, disassemble=args.dis)
 
 
-if not args.dont_wrap_pytest:
+if sys.version_info[0:2] < (3,12) and not args.dont_wrap_pytest:
     sc.wrap_pytest(sci, file_matcher)
 
 

src/slipcover/branch.py

@@ -4,6 +4,20 @@
 BRANCH_NAME = "_slipcover_branches"
 PYTHON_VERSION = sys.version_info[0:2]
 
+if sys.version_info[0:2] >= (3,12):
+    def is_branch(line):
+        return (line & (1<<30)) != 0
+
+    def encode_branch(from_line, to_line):
+        # FIXME anything bigger, and we get an overflow... encode to_line as relative number?
+        assert from_line <= 0x7FFF
+        assert to_line <= 0x7FFF
+        return (1<<30)|((from_line & 0x7FFF)<<15)|(to_line&0x7FFF)
+
+    def decode_branch(line):
+        return ((line>>15)&0x7FFF, line&0x7FFF)
+
+
 def preinstrument(tree: ast.AST) -> ast.AST:
     """Prepares an AST for Slipcover instrumentation, inserting assignments indicating where branches happen."""
 
@@ -15,9 +29,15 @@ def _mark_branch(self, from_line: int, to_line: int) -> ast.AST:
             mark = ast.Assign([ast.Name(BRANCH_NAME, ast.Store())],
                                ast.Tuple([ast.Constant(from_line), ast.Constant(to_line)], ast.Load()))
 
-            for node in ast.walk(mark):
-                # we ignore line 0, so this avoids generating extra line probes
-                node.lineno = 0 if PYTHON_VERSION >= (3,11) else from_line
+            if PYTHON_VERSION == (3,12):
+                for node in ast.walk(mark):
+                    node.lineno = node.end_lineno = encode_branch(from_line, to_line)
+            elif PYTHON_VERSION == (3,11):
+                for node in ast.walk(mark):
+                    node.lineno = 0 # we ignore line 0, so this avoids generating extra line probes
+            else:
+                for node in ast.walk(mark):
+                    node.lineno = from_line
 
             return [mark]
 

src/slipcover/importer.py

@@ -1,7 +1,6 @@
 from typing import Any
 from .slipcover import Slipcover, VERSION
 from . import branch as br
-from . import bytecode as bc
 from pathlib import Path
 import sys
 
@@ -93,13 +92,15 @@ def matches(self, filename : Path):
 
         return self.cwd in filename.parents
 
+
 class MatchEverything:
     def __init__(self):
         pass
 
     def matches(self, filename : Path):
         return True
 
+
 class SlipcoverMetaPathFinder(MetaPathFinder):
     def __init__(self, sci, file_matcher, debug=False):
         self.debug = debug
@@ -156,6 +157,8 @@ def __exit__(self, *args: Any) -> None:
 
 
 def wrap_pytest(sci: Slipcover, file_matcher: FileMatcher):
+    from . import bytecode as bc
+
     def exec_wrapper(obj, g):
         if hasattr(obj, 'co_filename') and file_matcher.matches(obj.co_filename):
             obj = sci.instrument(obj)

src/slipcover/slipcover.py

@@ -5,10 +5,13 @@
 from typing import Dict, Set, List
 from collections import defaultdict, Counter
 import threading
-from . import probe
-from . import bytecode as bc
-from . import branch as br
+
+if sys.version_info[0:2] < (3,12):
+    from . import probe
+    from . import bytecode as bc
+
 from pathlib import Path
+from . import branch as br
 
 VERSION = "0.3.2"
 
@@ -32,7 +35,6 @@ def simplify(self, path : str) -> str:
         except ValueError:
             return path 
 
-
 class Slipcover:
     def __init__(self, immediate: bool = False,
                  d_miss_threshold: int = 50, branch: bool = False, skip_covered: bool = False,
@@ -46,23 +48,36 @@ def __init__(self, immediate: bool = False,
         # mutex protecting this state
         self.lock = threading.RLock()
 
-        # maps to guide CodeType replacements
-        self.replace_map: Dict[types.CodeType, types.CodeType] = dict()
-        self.instrumented: Dict[str, set] = defaultdict(set)
-
         # notes which code lines have been instrumented
         self.code_lines: Dict[str, set] = defaultdict(set)
         self.code_branches: Dict[str, set] = defaultdict(set)
 
-        # provides an index (line_or_branch -> offset) for each code object
-        self.code2index: Dict[types.CodeType, list] = dict()
-
         # notes which lines and branches have been seen.
         self.all_seen: Dict[str, set] = defaultdict(set)
 
         # notes lines/branches seen since last de-instrumentation
         self._get_newly_seen()
 
+        if sys.version_info[0:2] >= (3,12):
+            def handle_line(code, line):
+                if br.is_branch(line):
+                    self.newly_seen[code.co_filename].add(br.decode_branch(line))
+                elif line:
+                    self.newly_seen[code.co_filename].add(line)
+                return sys.monitoring.DISABLE
+
+            if sys.monitoring.get_tool(sys.monitoring.COVERAGE_ID) != "SlipCover":
+                sys.monitoring.use_tool_id(sys.monitoring.COVERAGE_ID, "SlipCover") # FIXME add free_tool_id
+            sys.monitoring.register_callback(sys.monitoring.COVERAGE_ID,
+                                             sys.monitoring.events.LINE, handle_line)
+        else:
+            # maps to guide CodeType replacements
+            self.replace_map: Dict[types.CodeType, types.CodeType] = dict()
+            self.instrumented: Dict[str, set] = defaultdict(set)
+
+            # provides an index (line_or_branch -> offset) for each code object
+            self.code2index: Dict[types.CodeType, list] = dict()
+
         self.modules = []
 
     def _get_newly_seen(self):
@@ -80,99 +95,132 @@ def _get_newly_seen(self):
         return newly_seen
 
 
-    def instrument(self, co: types.CodeType, parent: types.CodeType = 0) -> types.CodeType:
-        """Instruments a code object for coverage detection.
+    if sys.version_info[0:2] >= (3,12):
+        def instrument(self, co: types.CodeType, parent: types.CodeType = 0) -> types.CodeType:
+            """Instruments a code object for coverage detection.
 
-        If invoked on a function, instruments its code.
-        """
+            If invoked on a function, instruments its code.
+            """
 
-        if isinstance(co, types.FunctionType):
-            co.__code__ = self.instrument(co.__code__)
-            return co.__code__
+            if isinstance(co, types.FunctionType):
+                co = co.__code__
 
-        assert isinstance(co, types.CodeType)
-        # print(f"instrumenting {co.co_name}")
+            assert isinstance(co, types.CodeType)
+            # print(f"instrumenting {co.co_name}")
 
-        ed = bc.Editor(co)
+            sys.monitoring.set_local_events(sys.monitoring.COVERAGE_ID, co, sys.monitoring.events.LINE)
 
-        # handle functions-within-functions
-        for i, c in enumerate(co.co_consts):
-            if isinstance(c, types.CodeType):
-                ed.set_const(i, self.instrument(c, co))
-
-        probe_signal_index = ed.add_const(probe.signal)
-
-        off_list = list(dis.findlinestarts(co))
-        if self.branch:
-            off_list.extend(list(ed.find_const_assignments(br.BRANCH_NAME)))
-            # sort line probes (2-tuples) before branch probes (3-tuples) because
-            # line probes don't overwrite bytecode like branch probes do, so if there
-            # are two being inserted at the same offset, the accumulated offset 'delta' applies
-            off_list.sort(key = lambda x: (x[0], len(x)))
-
-        branch_set = set()
-        insert_labels = []
-        probes = []
-
-        delta = 0
-        for off_item in off_list:
-            if len(off_item) == 2: # from findlinestarts
-                offset, lineno = off_item
-                if lineno == 0 or co.co_code[offset] == bc.op_RESUME:
-                    continue
+            # handle functions-within-functions
+            for c in co.co_consts:
+                if isinstance(c, types.CodeType):
+                    self.instrument(c, co)
 
-                # Can't insert between an EXTENDED_ARG and the final opcode
-                if (offset >= 2 and co.co_code[offset-2] == bc.op_EXTENDED_ARG):
-                    while (offset < len(co.co_code) and co.co_code[offset-2] == bc.op_EXTENDED_ARG):
-                        offset += 2 # TODO will we overtake the next offset from findlinestarts?
+            op_RESUME = dis.opmap["RESUME"]
 
-                insert_labels.append(lineno)
+            with self.lock:
+                # Python 3.11 generates a 0th line; 3.11+ generates a line just for RESUME
+                self.code_lines[co.co_filename].update(line for off, line in dis.findlinestarts(co) \
+                                                       if line != 0 and not br.is_branch(line) \
+                                                          and co.co_code[off] != op_RESUME)
 
-                tr = probe.new(self, co.co_filename, lineno, self.d_miss_threshold)
-                probes.append(tr)
-                tr_index = ed.add_const(tr)
+                self.code_branches[co.co_filename].update(br.decode_branch(line) for off, line in dis.findlinestarts(co) \
+                                                          if br.is_branch(line) and co.co_code[off] != op_RESUME)
+            return co
 
-                delta += ed.insert_function_call(offset+delta, probe_signal_index, (tr_index,))
+    else:
+        def instrument(self, co: types.CodeType, parent: types.CodeType = 0) -> types.CodeType:
+            """Instruments a code object for coverage detection.
 
-            else: # from find_const_assignments
-                begin_off, end_off, branch_index = off_item
-                branch = co.co_consts[branch_index]
+            If invoked on a function, instruments its code.
+            """
 
-                branch_set.add(branch)
-                insert_labels.append(branch)
+            if isinstance(co, types.FunctionType):
+                co.__code__ = self.instrument(co.__code__)
+                return co.__code__
 
-                tr = probe.new(self, co.co_filename, branch, self.d_miss_threshold)
-                probes.append(tr)
-                ed.set_const(branch_index, tr)
+            assert isinstance(co, types.CodeType)
+            # print(f"instrumenting {co.co_name}")
 
-                delta += ed.insert_function_call(begin_off+delta, probe_signal_index, (branch_index,),
-                                                 repl_length = end_off-begin_off)
+            ed = bc.Editor(co)
 
-        ed.add_const('__slipcover__')  # mark instrumented
-        new_code = ed.finish()
+            # handle functions-within-functions
+            for i, c in enumerate(co.co_consts):
+                if isinstance(c, types.CodeType):
+                    ed.set_const(i, self.instrument(c, co))
 
-        if self.disassemble:
-            dis.dis(new_code)
+            probe_signal_index = ed.add_const(probe.signal)
 
-        if self.immediate:
-            for tr, off in zip(probes, ed.get_inserts()):
-                probe.set_immediate(tr, new_code.co_code, off)
-        else:
-            index = list(zip(ed.get_inserts(), insert_labels))
+            off_list = list(dis.findlinestarts(co))
+            if self.branch:
+                off_list.extend(list(ed.find_const_assignments(br.BRANCH_NAME)))
+                # sort line probes (2-tuples) before branch probes (3-tuples) because
+                # line probes don't overwrite bytecode like branch probes do, so if there
+                # are two being inserted at the same offset, the accumulated offset 'delta' applies
+                off_list.sort(key = lambda x: (x[0], len(x)))
 
-        with self.lock:
-            # Python 3.11 generates a 0th line; 3.11+ generates a line just for RESUME
-            self.code_lines[co.co_filename].update(line for off, line in dis.findlinestarts(co) \
-                                                   if line != 0 and co.co_code[off] != bc.op_RESUME)
-            self.code_branches[co.co_filename].update(branch_set)
+            branch_set = set()
+            insert_labels = []
+            probes = []
 
-            if not parent:
-                self.instrumented[co.co_filename].add(new_code)
+            delta = 0
+            for off_item in off_list:
+                if len(off_item) == 2: # from findlinestarts
+                    offset, lineno = off_item
+                    if lineno == 0 or co.co_code[offset] == bc.op_RESUME:
+                        continue
 
-            if not self.immediate:
-                self.code2index[new_code] = index
+                    # Can't insert between an EXTENDED_ARG and the final opcode
+                    if (offset >= 2 and co.co_code[offset-2] == bc.op_EXTENDED_ARG):
+                        while (offset < len(co.co_code) and co.co_code[offset-2] == bc.op_EXTENDED_ARG):
+                            offset += 2 # TODO will we overtake the next offset from findlinestarts?
 
-        return new_code
+                    insert_labels.append(lineno)
+
+                    tr = probe.new(self, co.co_filename, lineno, self.d_miss_threshold)
+                    probes.append(tr)
+                    tr_index = ed.add_const(tr)
+
+                    delta += ed.insert_function_call(offset+delta, probe_signal_index, (tr_index,))
+
+                else: # from find_const_assignments
+                    begin_off, end_off, branch_index = off_item
+                    branch = co.co_consts[branch_index]
+
+                    branch_set.add(branch)
+                    insert_labels.append(branch)
+
+                    tr = probe.new(self, co.co_filename, branch, self.d_miss_threshold)
+                    probes.append(tr)
+                    ed.set_const(branch_index, tr)
+
+                    delta += ed.insert_function_call(begin_off+delta, probe_signal_index, (branch_index,),
+                                                     repl_length = end_off-begin_off)
+
+            ed.add_const('__slipcover__')  # mark instrumented
+            new_code = ed.finish()
+
+            if self.disassemble:
+                dis.dis(new_code)
+
+            if self.immediate:
+                for tr, off in zip(probes, ed.get_inserts()):
+                    probe.set_immediate(tr, new_code.co_code, off)
+            else:
+                index = list(zip(ed.get_inserts(), insert_labels))
+
+            with self.lock:
+                # Python 3.11 generates a 0th line; 3.11+ generates a line just for RESUME
+                self.code_lines[co.co_filename].update(line for off, line in dis.findlinestarts(co) \
+                                                       if line != 0 and co.co_code[off] != bc.op_RESUME)
+                self.code_branches[co.co_filename].update(branch_set)
+
+                if not parent:
+                    self.instrumented[co.co_filename].add(new_code)
+
+                if not self.immediate:
+                    self.code2index[new_code] = index
+
+            return new_code
 
 
     def deinstrument(self, co, lines: set) -> types.CodeType:

tests/test_bytecode.py

@@ -1,13 +1,16 @@
 import pytest
+import sys
+
+PYTHON_VERSION = sys.version_info[0:2]
+
+if PYTHON_VERSION >= (3,12):
+    pytest.skip(allow_module_level=True)
+
 import slipcover.bytecode as bc
 import types
 import dis
-import sys
 import inspect
 
-
-PYTHON_VERSION = sys.version_info[0:2]
-
 def current_line():
     import inspect as i
     return i.getframeinfo(i.currentframe().f_back).lineno