Async functions get synchronous trampolines causing "coroutine was never awaited" errors

[DamienGR]

Description

When using mutmut 3.4.0 with async functions (async def), the generated trampoline wrapper is synchronous (def), which causes the coroutine to not be awaited.

Environment

• mutmut version: 3.4.0
• Python version: 3.12.3
• OS: Linux (WSL2)

Steps to Reproduce

1. Have a codebase with async functions like:

async def get_current_user(
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    """Get current authenticated user from JWT token."""
    # ... async code
    user = await service.get_user_by_id(uuid.UUID(user_id))
    return user

2. Run mutmut run

3. Observe the generated trampoline in mutants/ directory

Expected Behavior

The trampoline for async functions should be:

async def get_current_user(*args, **kwargs):
    result = await _mutmut_trampoline(x_get_current_user__mutmut_orig, x_get_current_user__mutmut_mutants, args, kwargs)
    return result

Actual Behavior

The generated trampoline is synchronous:

def get_current_user(*args, **kwargs):
    result = _mutmut_trampoline(x_get_current_user__mutmut_orig, x_get_current_user__mutmut_mutants, args, kwargs)
    return result 

This causes the error:

RuntimeWarning: coroutine 'x_get_current_user__mutmut_orig' was never awaited

And tests fail because the coroutine is returned instead of being awaited.

Root Cause

In trampoline_templates.py, the build_trampoline function always generates a synchronous def:

def {orig_name}({'self, ' if class_name is not None else ''}*args, **kwargs):
    result = {trampoline_name}(...)
    return result 

It should detect if the original function is async and generate:

async def {orig_name}({'self, ' if class_name is not None else ''}*args, **kwargs):
    result = await {trampoline_name}(...)
    return result 

Related

• #406 - Fixed async generators, but not regular async functions
• #407 - PR that added async generator support

Workaround

Currently excluding files with async functions using do_not_mutate in pyproject.toml:

[tool.mutmut]
do_not_mutate = [
    "**/dependencies.py",
    "**/routes.py", 
    "**/service.py",
    "**/database.py",
]

This significantly limits mutation testing coverage for async Python codebases (FastAPI, asyncio, etc.).

[Otto-AA]

Hi, at least in theory async await is already supported. Can you provide a minimal reproducible example for this?

We also have an e2e test for this that works as expected:

https://github.com/boxed/mutmut/blob/3ca1d44811636f0a39801be88adafe7ec05b6853/e2e_projects/my_lib/tests/test_my_lib.py#L43-L46

https://github.com/boxed/mutmut/blob/3ca1d44811636f0a39801be88adafe7ec05b6853/e2e_projects/my_lib/src/my_lib/init.py#L31-L40

The generated trampoline is synchronous: This causes the error: RuntimeWarning: coroutine 'x_get_current_user__mutmut_orig' was never awaited

If I'm not mistaken, the coroutine does not need to be awaited by the trampoline. If the trampoline simply returns the coroutine object, the caller of the mutated method can await it and then I don't think this RuntimeWarning occurs. I may be wrong or forgetting something, but hard to tell without a minimal repro.

[DamienGR]

I tested PR #455 from @Julien-Delavisse and can confirm it fixes the async function issue for our FastAPI codebase.

Minimal Reproducible Example

Here's a minimal example that demonstrates the issue (fails with main branch, passes with PR #455):

async_example.py:

async def async_add(a: int, b: int) -> int:
    return a + b

test_async.py:

import pytest
from async_example import async_add

@pytest.mark.asyncio
async def test_async_add():
    result = await async_add(1, 2)
    assert result == 3

pyproject.toml:

[tool.mutmut]
paths_to_mutate = ["async_example.py"]
also_copy = ["."]

Results

With the main branch (pip install mutmut):

• Test collection fails or produces RuntimeWarning: coroutine 'async_add__mutmut_orig' was never awaited

With PR #455 (pip install git+https://github.com/Julien-Delavisse/mutmut.git@fix/454-async-functions-generated-trampoline-wrapper-is-synchronous):

• Mutation testing runs successfully: 1/1 mutants killed

Real-world validation

I also tested PR #455 on our FastAPI async service layer (auth/service.py with ~20 async methods) and mutation testing completed successfully without any "coroutine was never awaited" errors.

The fix in PR #455 correctly adds async/await keywords to the generated trampolines for async functions, which is necessary for pytest-asyncio to properly handle the coroutines.

[Julien-Delavisse]

Hi @DamienGR

I reformatted your Minimal Reproducible Example and it works with the current version of mutmut, without the PR. https://github.com/Julien-Delavisse/mutmut-tests

Can you see if you can modify it enough to make the problem appear?

[DamienGR]

@Julien-Delavisse Thanks for testing! I apologize - I tried to create several minimal reproducible examples but none of them reproduce the issue. The problem persists in our real codebase but I cannot isolate it to a minimal example yet.

I'll continue investigating on my side to identify what specific combination of factors triggers the issue in our project. Once I find a proper reproducible case, I'll update this issue.

Sorry for the noise, and thanks for your patience!

[DamienGR]

Minimal Reproducible Example Found!

I've isolated the issue. The root cause is that asyncio.iscoroutinefunction() returns False for trampolined async functions.

The Core Problem

Original code:

async def get_value() -> int:
    await asyncio.sleep(0)
    return 42

Generated trampoline:

def get_value(*args, **kwargs):  # ← SYNC function!
    result = _mutmut_trampoline(...)
    return result

Result:

asyncio.iscoroutinefunction(get_value)  # Returns False instead of True!

Why This Breaks Frameworks

Frameworks like FastAPI, Starlette, and others use iscoroutinefunction() to decide whether to await a function. When it returns False, they don't await → "coroutine was never awaited".

Minimal Reproduction (4 files, no external dependencies)

src/mymod/async_func.py:

import asyncio

async def get_value() -> int:
    await asyncio.sleep(0)
    return 42

tests/test_async.py:

import asyncio

def test_iscoroutinefunction():
    from mymod.async_func import get_value
    assert asyncio.iscoroutinefunction(get_value), (
        "get_value should be detected as coroutine function"
    )

pyproject.toml:

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "mutmut-minimal-bug"
version = "0.1.0"
requires-python = ">=3.12"

[project.optional-dependencies]
dev = ["pytest>=7.4.0", "mutmut>=3.0.0"]

[tool.setuptools.packages.find]
where = ["src"]

[tool.mutmut]
paths_to_mutate = ["src/"]
pytest_add_cli_args_test_selection = ["tests/"]

src/mymod/__init__.py: (empty file) tests/__init__.py: (empty file)

Steps to Reproduce

# Setup
python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

# Normal test passes
pytest tests/ -v  # ✅ PASSED

# Mutmut fails
mutmut run  # ❌ FAILED - "get_value should be detected as coroutine function"

Why Previous Reproductions Didn't Work

Simple await async_func() tests pass because the trampoline returns a coroutine object that gets awaited by the caller. The issue only manifests when code inspects the function to decide whether to await (which is what FastAPI does with dependencies).

Environment

• mutmut: 3.4.0 (official PyPI)
• Python: 3.12.3
• OS: Linux

[Otto-AA]

Ah yes, making it sync changes the function signature, and fastapi uses the signature to decide what to do. So we should try to keep the same signature, if possible.

Thanks for the example, now it makes sense to me!

[DamienGR]

Sorry I didn't intend to close this issue !