Adds Python 3.12 support using sys.monitoring

.github/workflows/tests.yml

@@ -19,7 +19,7 @@ jobs:
     strategy:
       matrix:
         os: [ ubuntu-latest, macos-latest, windows-latest ]
-        python: [ '3.8', '3.9', 'pypy-3.9', '3.10', '3.11' ]
+        python: [ '3.8', '3.9', 'pypy-3.9', '3.10', '3.11', '3.12' ]
 
     steps:
     - uses: actions/checkout@v3
@@ -71,7 +71,7 @@ jobs:
     container: ${{ matrix.container }}
     strategy:
       matrix:
-        python_version: ['3.8', '3.9', '3.10', '3.11']
+        python_version: ['3.8', '3.9', '3.10', '3.11', '3.12']
         include:
           - os: ubuntu-latest
             container: quay.io/pypa/manylinux2014_x86_64 # https://github.com/pypa/manylinux

Makefile

@@ -5,7 +5,7 @@ all:
 HOMEBREW_PYTHON=/opt/homebrew/opt/python@
 test:
 	- rm -f .coverage
-	@ for V in 3.8 3.9 3.10 3.11; do \
+	@ for V in 3.8 3.9 3.10 3.11 3.12; do \
 	    P=$$(command -v ${HOMEBREW_PYTHON}$$V/bin/python3 || command -v python$$V); \
 	    if ! [ -z $$P ]; then \
 	      $$P --version; \

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/branch.py

@@ -4,6 +4,20 @@
 
 BRANCH_NAME = "_slipcover_branches"
 
+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, f"Line number {from_line} too high, unable to add branch tracking"
+        assert to_line <= 0x7FFF, f"Line number {to_line} too high, unable to add branch tracking"
+        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) -> List[ast.stmt]:
             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 sys.version_info >= (3,11) else from_line
+            if sys.version_info[0:2] == (3,12):
+                for node in ast.walk(mark):
+                    node.lineno = node.end_lineno = encode_branch(from_line, to_line)
+            elif sys.version_info[0:2] == (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/bytecode.py

@@ -590,56 +590,6 @@ def disable_inserted_function(self, offset):
         self.patch[offset] = op_JUMP_FORWARD
 
 
-    def replace_global_with_const(self, global_name, const_index):
-        """Replaces a global name lookup by a constant load."""
-        assert not self.finished
-
-        if self.patch is None:
-            self.patch = bytearray(self.orig_code.co_code)
-
-        if self.branches is None:
-            self.branches = Branch.from_code(self.orig_code)
-            self.ex_table = ExceptionTableEntry.from_code(self.orig_code)
-            self.lines = LineEntry.from_code(self.orig_code)
-
-        if global_name in self.orig_code.co_names:
-            name_index = self.orig_code.co_names.index(global_name)
-
-            def find_load_globals():
-                for op_off, op_len, op, op_arg in unpack_opargs(self.patch):
-                    if op == op_LOAD_GLOBAL:
-                        if sys.version_info >= (3,11):
-                            if (op_arg>>1) == name_index:
-                                yield (op_off, op_len, op, op_arg)
-                        else:
-                            if op_arg == name_index:
-                                yield (op_off, op_len, op, op_arg)
-
-            delta = 0
-            # read from pre-computed list() below so we can modify on the fly
-            for op_off, op_len, op, op_arg in list(find_load_globals()):
-                repl = bytearray()
-                if sys.version_info[0:2] >= (3,11) and op_arg&1:
-                    repl.extend(opcode_arg(dis.opmap['PUSH_NULL'], 0))
-                repl.extend(opcode_arg(op_LOAD_CONST, const_index))
-
-                op_off += delta     # adjust for any other changes
-                self.patch[op_off:op_off+op_len] = repl
-
-                change = len(repl) - op_len
-                if change:
-                    for l in self.lines:
-                        l.adjust(op_off, change)
-
-                    for b in self.branches:
-                        b.adjust(op_off, change)
-
-                    for e in self.ex_table:
-                        e.adjust(op_off, change)
-
-                delta += change
-
-
     def _finish(self):
         if not self.finished:
             self.finished = True

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,22 +157,59 @@ def __exit__(self, *args: Any) -> None:
 
 
 def wrap_pytest(sci: Slipcover, file_matcher: FileMatcher):
-    def exec_wrapper(obj, g):
-        if hasattr(obj, 'co_filename') and file_matcher.matches(obj.co_filename):
-            obj = sci.instrument(obj)
-        exec(obj, g)
+    def redirect_calls(module, funcName, funcWrapperName):
+        """Redirects calls to the given function to a wrapper function in the same module."""
+        import ast
+        import types
+
+        assert funcWrapperName not in module.__dict__, f"function {funcWrapperName} already defined"
+
+        with open(module.__file__) as f:
+            t = ast.parse(f.read())
+
+        funcNames = set() # names of the functions we modified
+        for n in ast.walk(t):
+            if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                for s in ast.walk(n):
+                    if isinstance(s, ast.Call) and isinstance(s.func, ast.Name) and s.func.id == funcName:
+                        s.func.id = funcWrapperName
+                        funcNames.add(n.name)
+
+        code = compile(t, module.__file__, "exec")
+
+        # It's tempting to just exec(code, module.__dict__) here, but the code often times has side effects...
+        # So instead of we find the new code object(s) and replace them in the loaded module.
+
+        replacement = dict() # replacement code objects
+        def find_replacements(co):
+            for c in co.co_consts:
+                if isinstance(c, types.CodeType):
+                    if c.co_name in funcNames:
+                        replacement[c.co_name] = c
+                    find_replacements(c)
+
+        find_replacements(code)
+
+        visited = set()
+        for f in Slipcover.find_functions(module.__dict__.values(), visited):
+            if (repl := replacement.get(f.__code__.co_name, None)):
+                assert f.__code__.co_firstlineno == repl.co_firstlineno # sanity check
+                f.__code__ = repl
+
 
     try:
         import _pytest.assertion.rewrite as pyrewrite
     except ModuleNotFoundError:
         return
 
-    for f in Slipcover.find_functions(pyrewrite.__dict__.values(), set()):
-        if 'exec' in f.__code__.co_names:
-            ed = bc.Editor(f.__code__)
-            wrapper_index = ed.add_const(exec_wrapper)
-            ed.replace_global_with_const('exec', wrapper_index)
-            f.__code__ = ed.finish()
+    redirect_calls(pyrewrite, "exec", "_Slipcover_exec_wrapper")
+
+    def exec_wrapper(obj, g):
+        if hasattr(obj, 'co_filename') and file_matcher.matches(obj.co_filename):
+            obj = sci.instrument(obj)
+        exec(obj, g)
+
+    pyrewrite._Slipcover_exec_wrapper = exec_wrapper
 
     if sci.branch:
         import inspect