Preserve function signature

CONTRIBUTING.rst

@@ -21,7 +21,7 @@ Running the tests
 
     uv run pytest
 
-This also runs E2E tests that verify that `mutmut run` produces the same output as before. If your code changes should change the output of `mutmut run` and this test fails, try to delete the `snapshots/*.json` files (as described in the test errors).
+We use `inline-snapshot` for E2E and integration tests, to prevent unexpected changes in the output. If the output _should_ change, you can use `uv run pytest --inline-snapshot=fix` to update the snapshots.
 
 If pytest terminates before reporting the test failures, it likely hit a case where mutmut calls `os._exit(...)`. Try looking at these calls first for troubleshooting.
 

e2e_projects/my_lib/src/my_lib/__init__.py

@@ -104,9 +104,9 @@ def some_func(a, b: str = "111", c: Callable[[str], int] | None = None) -> int |
         return c(b)
     return None
 
-def func_with_star_clone(a, *, b, **kwargs): pass  # pragma: no mutate
-def func_with_star(a, *, b, **kwargs):
-    return a + b + len(kwargs)
+def func_with_star_clone(a, /, b, *, c, **kwargs): pass  # pragma: no mutate
+def func_with_star(a, /, b, *, c, **kwargs):
+    return a + b + c + len(kwargs)
 
 def func_with_arbitrary_args_clone(*args, **kwargs): pass  # pragma: no mutate
 def func_with_arbitrary_args(*args, **kwargs):

e2e_projects/my_lib/tests/test_my_lib.py

@@ -1,6 +1,7 @@
 import inspect
 from my_lib import *
 import pytest
+import asyncio
 
 """These tests are flawed on purpose, some mutants survive and some are killed."""
 
@@ -60,5 +61,9 @@ def test_that_signatures_are_preserved():
 
 def test_signature_functions_are_callable():
     assert some_func(True, c=lambda s: int(s), b="222") == 222
-    assert func_with_star(1, b=2, x='x', y='y', z='z') == 6
+    assert func_with_star(1, b=2, c=3, x='x', y='y', z='z') == 9
     assert func_with_arbitrary_args('a', 'b', foo=123, bar=456) == 4
+
+def test_signature_is_coroutine():
+    assert asyncio.iscoroutinefunction(async_consumer)
+

pyproject.toml

@@ -53,11 +53,16 @@ source-include = ["HISTORY.rst"]
 
 [dependency-groups]
 dev = [
+    "inline-snapshot>=0.32.0",
     "pytest-asyncio>=1.0.0",
+    "ruff>=0.15.1",
 ]
 
 [tool.pytest.ini_options]
 testpaths = [
     "tests",
 ]
 asyncio_default_fixture_loop_scope = "function"
+
+[tool.inline-snapshot]
+format-command="ruff format --stdin-filename {filename}"

src/mutmut/code_coverage.py

@@ -2,7 +2,6 @@
 import importlib
 import sys
 from pathlib import Path
-import json
 
 
 # Returns a set of lines that are covered in this file gvein the covered_lines dict

src/mutmut/file_mutation.py

@@ -3,13 +3,11 @@
 from collections import defaultdict
 from collections.abc import Iterable, Sequence, Mapping
 from dataclasses import dataclass
-from pathlib import Path
 from typing import Union
-import warnings
 import libcst as cst
 from libcst.metadata import PositionProvider, MetadataWrapper
 import libcst.matchers as m
-from mutmut.trampoline_templates import build_trampoline, mangle_function_name, trampoline_impl
+from mutmut.trampoline_templates import create_trampoline_lookup, mangle_function_name, trampoline_impl
 from mutmut.node_mutation import mutation_operators, OPERATORS_TYPE
 
 NEVER_MUTATE_FUNCTION_NAMES = { "__getattribute__", "__setattr__", "__new__" }
@@ -248,13 +246,88 @@ def function_trampoline_arrangement(function: cst.FunctionDef, mutants: Iterable
         nodes.append(mutated_method) # type: ignore
 
     # trampoline that forwards the calls
-    trampoline = list(cst.parse_module(build_trampoline(orig_name=name, mutants=mutant_names, class_name=class_name)).body)
-    trampoline[0] = trampoline[0].with_changes(leading_lines=[cst.EmptyLine()])
-    nodes.extend(trampoline)
+    trampoline = create_trampoline_wrapper(function, mangled_name, class_name)
+    mutants_dict = list(cst.parse_module(create_trampoline_lookup(orig_name=name, mutants=mutant_names, class_name=class_name)).body)
+    mutants_dict[0] = mutants_dict[0].with_changes(leading_lines=[cst.EmptyLine()])
+
+    nodes.append(trampoline)
+    nodes.extend(mutants_dict)
 
     return nodes, mutant_names
 
 
+def create_trampoline_wrapper(function: cst.FunctionDef, mangled_name: str, class_name: str | None) -> cst.FunctionDef:
+    args: list[cst.Element | cst.StarredElement] = []
+    for pos_only_param in function.params.posonly_params:
+        args.append(cst.Element(pos_only_param.name))
+    for param in function.params.params:
+        args.append(cst.Element(param.name))
+    if isinstance(function.params.star_arg, cst.Param):
+        args.append(cst.StarredElement(function.params.star_arg.name))
+
+    if class_name is not None:
+        # remove self arg (handled by the trampoline function)
+        args = args[1:]
+
+    args_assignemnt = cst.Assign([cst.AssignTarget(cst.Name(value='args'))], cst.List(args))
+
+    kwargs: list[cst.DictElement | cst.StarredDictElement] = []
+    for param in function.params.kwonly_params:
+        kwargs.append(cst.DictElement(cst.SimpleString(f"'{param.name.value}'"), param.name))
+    if isinstance(function.params.star_kwarg, cst.Param):
+        kwargs.append(cst.StarredDictElement(function.params.star_kwarg.name))
+
+    kwargs_assignment = cst.Assign([cst.AssignTarget(cst.Name(value='kwargs'))], cst.Dict(kwargs))
+
+    def _get_local_name(func_name: str) -> cst.BaseExpression:
+        # for top level, simply return the name
+        if class_name is None:
+            return cst.Name(func_name)
+        # for class methods, use object.__getattribute__(self, name)
+        return cst.Call(
+            func=cst.Attribute(cst.Name('object'), cst.Name('__getattribute__')),
+            args=[cst.Arg(cst.Name('self')), cst.Arg(cst.SimpleString(f"'{func_name}'"))]
+        )
+
+    result: cst.BaseExpression = cst.Call(
+        func=cst.Name('_mutmut_trampoline'),
+        args=[
+            cst.Arg(_get_local_name(f'{mangled_name}_orig')),
+            cst.Arg(_get_local_name(f'{mangled_name}_mutants')),
+            cst.Arg(cst.Name('args')),
+            cst.Arg(cst.Name('kwargs')),
+            cst.Arg(cst.Name('None' if class_name is None else 'self')),
+        ],
+    )
+    # for non-async functions, simply return the value or generator
+    result_statement = cst.SimpleStatementLine([cst.Return(result)])
+
+    if function.asynchronous:
+        is_generator = _is_generator(function)
+        if is_generator:
+            # async for i in _mutmut_trampoline(...): yield i
+            result_statement = cst.For(
+                target=cst.Name('i'),
+                iter=result,
+                body=cst.IndentedBlock([cst.SimpleStatementLine([cst.Expr(cst.Yield(cst.Name('i')))])]),
+                asynchronous=cst.Asynchronous(),
+            )
+        else:
+            # return await _mutmut_trampoline(...)
+            result_statement = cst.SimpleStatementLine([cst.Return(cst.Await(result))])
+
+    function.whitespace_after_type_parameters
+    return function.with_changes(
+        body=cst.IndentedBlock(
+            [
+                cst.SimpleStatementLine([args_assignemnt]),
+                cst.SimpleStatementLine([kwargs_assignment]),
+                result_statement,
+            ],
+        ),
+    )
+
+
 def get_statements_until_func_or_class(statements: Sequence[MODULE_STATEMENT]) -> list[MODULE_STATEMENT]:
     """Get all statements until we encounter the first function or class definition"""
     result = []
@@ -302,3 +375,25 @@ def on_leave(self, original_node: cst.CSTNode, updated_node: cst.CSTNode) -> cst
             self.replaced_node = True
             return self.new_node
         return updated_node
+
+def _is_generator(function: cst.FunctionDef) -> bool:
+    """Return True if the function has yield statement(s)."""
+    visitor = IsGeneratorVisitor(function)
+    function.visit(visitor)
+    return visitor.is_generator
+
+class IsGeneratorVisitor(cst.CSTVisitor):
+    """Check if a function is a generator.
+    We do so by checking if any child is a Yield statement, but not looking into inner function definitions."""
+    def __init__(self, original_function: cst.FunctionDef):
+        self.is_generator = False
+        self.original_function: cst.FunctionDef = original_function
+
+    def visit_FunctionDef(self, node):
+        # do not recurse into inner function definitions
+        if self.original_function != node:
+            return False
+
+    def visit_Yield(self, node):
+        self.is_generator = True
+        return False

src/mutmut/trampoline_templates.py

@@ -1,27 +1,11 @@
 CLASS_NAME_SEPARATOR = 'ǁ'
 
-def build_trampoline(*, orig_name, mutants, class_name):
+def create_trampoline_lookup(*, orig_name, mutants, class_name):
     mangled_name = mangle_function_name(name=orig_name, class_name=class_name)
 
     mutants_dict = f'{mangled_name}__mutmut_mutants : ClassVar[MutantDict] = {{\n' + ', \n    '.join(f'{repr(m)}: {m}' for m in mutants) + '\n}'
-    access_prefix = ''
-    access_suffix = ''
-    self_arg = ''
-    if class_name is not None:
-        access_prefix = f'object.__getattribute__(self, "'
-        access_suffix = '")'
-        self_arg = ', self'
-
-    trampoline_name = '_mutmut_trampoline'
-
     return f"""
 {mutants_dict}
-
-def {orig_name}({'self, ' if class_name is not None else ''}*args, **kwargs):
-    result = {trampoline_name}({access_prefix}{mangled_name}__mutmut_orig{access_suffix}, {access_prefix}{mangled_name}__mutmut_mutants{access_suffix}, args, kwargs{self_arg})
-    return result 
-
-_mutmut_copy_signature({orig_name}, {mangled_name}__mutmut_orig)
 {mangled_name}__mutmut_orig.__name__ = '{mangled_name}'
 """
 
@@ -37,31 +21,10 @@ def mangle_function_name(*, name, class_name):
 # noinspection PyUnresolvedReferences
 # language=python
 trampoline_impl = """
-import inspect as _mutmut_inspect
-import sys as _mutmut_sys
 from typing import Annotated
 from typing import Callable
 from typing import ClassVar
 
-def _mutmut_copy_signature(trampoline, original_method):
-    if _mutmut_sys.version_info >= (3, 14):
-        # PEP 649 introduced deferred loading for annotations
-        # When some_method.__annotations__ is accessed, Python uses some_method.__annotate__(format) to compute it
-        # By copying the original __annotate__ method, we provide get the original annotations
-        trampoline.__annotate__ = original_method.__annotate__
-    else:
-        # On Python 3.13 and earlier, __annotations__ can be accessed immediately
-        trampoline.__annotations__ = original_method.__annotations__
-
-    try:
-        trampoline.__signature__ = _mutmut_inspect.signature(original_method)
-    except NameError:
-        # Also, because of PEP 649, it can happen that we cannot eagerly evaluate the signature
-        # In this case, fall back to stringifying the signature (which could cause different behaviour with runtime introspection)
-        trampoline.__signature__ = _mutmut_inspect.signature(original_method, annotation_format=_mutmut_inspect.Format.STRING)
-
-
-
 MutantDict = Annotated[dict[str, Callable], "Mutant"]
 
 
@@ -75,6 +38,7 @@ def _mutmut_trampoline(orig, mutants, call_args, call_kwargs, self_arg = None):
     elif mutant_under_test == 'stats':
         from mutmut.__main__ import record_trampoline_hit
         record_trampoline_hit(orig.__module__ + '.' + orig.__name__)
+        # (for class methods, orig is bound and thus does not need the explicit self argument)
         result = orig(*call_args, **call_kwargs)
         return result
     prefix = orig.__module__ + '.' + orig.__name__ + '__mutmut_'

tests/e2e/test_e2e_result_snapshots.py → tests/e2e/e2e_utils.py

@@ -4,8 +4,6 @@
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
-import pytest
-import sys
 
 import mutmut
 from mutmut.__main__ import SourceFileMutationData, _run, ensure_config_loaded, walk_source_files
@@ -47,8 +45,10 @@ def write_json_file(path: Path, data: Any):
         json.dump(data, file, indent=2)
 
 
-def asserts_results_did_not_change(project: str):
+def run_mutmut_on_project(project: str) -> dict:
     """Runs mutmut on this project and verifies that the results stay the same for all mutations."""
+    mutmut._reset_globals()
+
     project_path = Path("..").parent / "e2e_projects" / project
 
     mutants_path = project_path / "mutants"
@@ -58,37 +58,4 @@ def asserts_results_did_not_change(project: str):
     with change_cwd(project_path):
         _run([], None)
 
-    results = read_all_stats_for_project(project_path)
-
-    snapshot_path = Path("tests") / "e2e" / "snapshots" / (project + ".json")
-
-    if snapshot_path.exists():
-        # compare results against previous snapshot
-        previous_snapshot = read_json_file(snapshot_path)
-
-        err_msg = f'Mutmut results changed for the E2E project \'{project}\'. If this change was on purpose, delete {snapshot_path} and rerun the tests.'
-        assert results == previous_snapshot, err_msg
-    else:
-        # create the first snapshot
-        write_json_file(snapshot_path, results)
-
-
-def test_my_lib_result_snapshot():
-    mutmut._reset_globals()
-    asserts_results_did_not_change("my_lib")
-
-
-def test_config_result_snapshot():
-    mutmut._reset_globals()
-    asserts_results_did_not_change("config")
-
-
-def test_mutate_only_covered_lines_result_snapshot():
-    mutmut._reset_globals()
-    asserts_results_did_not_change("mutate_only_covered_lines")
-
-
-@pytest.mark.skipif(sys.version_info < (3, 14), reason="Can only test python 3.14 features on 3.14")
-def test_python_3_14_result_snapshot():
-    mutmut._reset_globals()
-    asserts_results_did_not_change("py3_14_features")
+    return read_all_stats_for_project(project_path)