Skip to content

It's not working with TestCase #9

@bfaust-bp

Description

@bfaust-bp

I have my tests in a TestCase, and pyleak isn't working with it.

# MagicMock subclass that supports async calls for mocking async functions
class AsyncMagicMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super().__call__(*args, **kwargs)

I asked the IA about the issue, and the answer was:

Looking at the pyleak pytest plugin source code, I can identify exactly why it doesn't work with TestCase-based tests:

The Issue

The key problem is in how pyleak wraps the test function:

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item: pytest.Function):
    """Wrap test execution with leak detection"""
    ...
    original_func = item.function
    ...
    if is_async:
        async def async_wrapper(*args, **kwargs):
            detector = CombinedLeakDetector(...)
            async with detector:
                return await original_func(*args, **kwargs)

        item.obj = async_wrapper  # <-- This is the problem!

The plugin replaces item.obj with the wrapped version, but for TestCase-based tests, this replacement is ignored.

Why It Doesn't Work with TestCase

For standalone async functions:

  • pytest directly calls item.obj()
  • The wrapper replacement works ✅

For TestCase methods (like IsolatedAsyncioTestCase):

  • pytest creates an instance of the TestCase class
  • The test method is called via IsolatedAsyncioTestCase._callTestMethod(self.<method_name>)
  • The method is retrieved from the instance's attributes (e.g., self.test_foo)
  • The item.obj replacement is completely bypassed ❌

In Python's unittest flow:

TestCase.run() → _callTestMethod(testMethod) → self._asyncioRunner.run(method)

The testMethod is the method bound to self, not item.obj. So pyleak's wrapper never gets executed.

Potential Fix (for pyleak)

To fix this, pyleak would need to patch the actual method on the TestCase instance:

if hasattr(item, 'instance') and item.instance is not None:
    # For TestCase methods, patch the method on the instance
    setattr(item.instance, item.name, types.MethodType(async_wrapper, item.instance))

This is a working version i'm using:

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item: pytest.Function) -> Any:  # type: ignore[type-arg]
    """
    Pytest hook to fix pyleak's no_leaks marker for IsolatedAsyncioTestCase methods.

    pyleak's pytest plugin replaces item.obj with a wrapped function, but for TestCase
    methods, the test is executed via IsolatedAsyncioTestCase._callTestMethod which
    retrieves the method from the instance, not from item.obj.

    This hook fixes that by patching the method directly on the TestCase instance.

    Reference: https://github.com/deepankarm/pyleak/blob/main/src/pyleak/pytest_plugin.py

    Args:
        item: The pytest test item being executed

    Yields:
        Control back to pytest for test execution

    Returns:
        None
    """
    # Import pyleak components for leak detection
    try:
        from pyleak.combined import CombinedLeakDetector, PyLeakConfig
        from pyleak.utils import CallerContext
    except ImportError:
        # pyleak not installed, skip
        yield
        return

    # Check if this test has the no_leaks marker
    marker = item.get_closest_marker("no_leaks")
    if not marker:
        yield
        return

    # Check if this is a TestCase-based test
    instance = getattr(item, "instance", None)
    if not isinstance(instance, TestCase):
        # Not a TestCase, let pyleak's default plugin handle it
        yield
        return

    # Check if the test method is async
    test_func = getattr(item, "obj", None) or getattr(item, "function", None)
    if test_func is None or not inspect.iscoroutinefunction(test_func):
        # Sync test, let pyleak's default plugin handle it
        yield
        return

    # Parse marker arguments (same logic as pyleak's should_monitor_test)
    marker_args: dict[str, Any] = {}
    if marker.args:
        for arg in marker.args:
            if arg == "tasks":
                marker_args["tasks"] = True
            elif arg == "threads":
                marker_args["threads"] = True
            elif arg == "blocking":
                marker_args["blocking"] = True
            elif arg == "all":
                marker_args.update({"tasks": True, "threads": True, "blocking": True})

    if marker.kwargs:
        marker_args.update(marker.kwargs)

    if not marker_args:
        marker_args = {"tasks": True, "threads": True, "blocking": True}

    # Create pyleak config
    config = PyLeakConfig.from_marker_args(marker_args)
    caller_context = CallerContext(
        filename=str(item.fspath) if item.fspath else "<unknown>",
        name=item.name,
        lineno=None,
    )

    # Get the original method from the instance (this is a bound method)
    original_method = getattr(instance, item.name)

    # Create wrapped async method with leak detection
    # Note: original_method is already bound, so we don't pass self again
    async def async_wrapper() -> Any:
        detector = CombinedLeakDetector(config=config, is_async=True, caller_context=caller_context)
        async with detector:
            return await original_method()

    # Patch the method on the instance so IsolatedAsyncioTestCase._callTestMethod picks it up
    # We use a simple function since test methods don't take args (except self which is already bound)
    setattr(instance, item.name, async_wrapper)

    try:
        yield
    finally:
        # Restore the original method
        setattr(instance, item.name, original_method)

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions