tessl/pypi-pyleak

Detect leaked asyncio tasks, threads, and event loop blocking in Python, inspired by Go's goleak package

—

Pending

Overview

Eval results

Files

PyTest Integration

Name: tessl/pypi-pyleak
Author: tessl

Automatic leak detection in test suites using pytest markers. The pyleak pytest plugin seamlessly integrates with existing test frameworks to provide comprehensive leak detection during testing without requiring manual context manager setup.

Capabilities

Pytest Plugin Registration

The plugin is automatically registered when pyleak is installed.

# Entry point configuration (in pyproject.toml):
[project.entry-points."pytest11"]
pyleak = "pyleak.pytest_plugin"

Test Marker

Primary marker for enabling leak detection in tests.

@pytest.mark.no_leaks  # Enable all leak detection types
@pytest.mark.no_leaks(tasks=True, threads=False, blocking=True)  # Selective detection
def test_function(): ...

@pytest.mark.no_leaks("tasks", "threads")  # Enable specific types
def test_function(): ...

@pytest.mark.no_leaks("all")  # Enable all types (equivalent to no arguments)
def test_function(): ...

Configuration Classes

Configuration class for customizing leak detection behavior.

class PyLeakConfig:
    """Configuration for pyleak detection."""
    
    # Task detection settings
    tasks: bool = True
    task_action: str = "raise"
    task_name_filter: str | None = None
    enable_task_creation_tracking: bool = False
    
    # Thread detection settings
    threads: bool = True
    thread_action: str = "raise"
    thread_name_filter: str | None = DEFAULT_THREAD_NAME_FILTER
    exclude_daemon_threads: bool = True
    
    # Event loop blocking detection settings
    blocking: bool = True  
    blocking_action: str = "raise"
    blocking_threshold: float = 0.2
    blocking_check_interval: float = 0.01
    
    @classmethod
    def from_marker_args(cls, marker_args: dict[str, Any]) -> "PyLeakConfig":
        """Create configuration from pytest marker arguments."""
    
    def to_markdown_table(self) -> str:
        """Generate markdown table of configuration options."""

Combined leak detector for coordinating multiple detection types.

class CombinedLeakDetector:
    """Combined detector for all leak types."""
    
    def __init__(
        self,
        config: PyLeakConfig,
        is_async: bool,
        caller_context: CallerContext | None = None,
    ): ...
    
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
    def __enter__(self): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...

Plugin Functions

Function to determine if a test should be monitored.

def should_monitor_test(item: pytest.Function) -> PyLeakConfig | None:
    """
    Check if test should be monitored and return config.
    
    Args:
        item: pytest Function item
        
    Returns:
        PyLeakConfig if test should be monitored, None otherwise
    """

Pytest hook for wrapping test execution.

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item: pytest.Function):
    """Wrap test execution with leak detection."""

Configuration Options

Marker Arguments

All configuration options available through the @pytest.mark.no_leaks marker:

Name	Default	Description
tasks	True	Whether to detect task leaks
task_action	raise	Action to take when a task leak is detected
task_name_filter	None	Filter to apply to task names
enable_task_creation_tracking	False	Whether to enable task creation tracking
threads	True	Whether to detect thread leaks
thread_action	raise	Action to take when a thread leak is detected
thread_name_filter	DEFAULT_THREAD_NAME_FILTER	Filter to apply to thread names (default: exclude asyncio threads)
exclude_daemon_threads	True	Whether to exclude daemon threads
blocking	True	Whether to detect event loop blocking
blocking_action	raise	Action to take when a blocking event loop is detected
blocking_threshold	0.2	Threshold for blocking event loop detection
blocking_check_interval	0.01	Interval for checking for blocking event loop

Usage Examples

Basic Setup

Add the marker to your pytest configuration:

pyproject.toml

[tool.pytest.ini_options]
markers = [
    "no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking"
]

pytest.ini

[tool:pytest]
markers = no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking

conftest.py

import pytest

def pytest_configure(config):
    config.addinivalue_line(
        "markers", 
        "no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking"
    )

Basic Test Detection

import pytest
import asyncio
import threading
import time

@pytest.mark.no_leaks
@pytest.mark.asyncio
async def test_async_no_leaks():
    # All leak types will be detected (tasks, threads, blocking)
    await asyncio.sleep(0.1)  # This is fine

@pytest.mark.no_leaks
def test_sync_no_leaks():
    # Only thread leaks will be detected (tasks and blocking require async context)
    time.sleep(0.1)  # This is fine

Selective Detection

# Only detect task leaks and event loop blocking
@pytest.mark.no_leaks(tasks=True, blocking=True, threads=False)
@pytest.mark.asyncio
async def test_selective_detection():
    asyncio.create_task(asyncio.sleep(10))  # This will be detected
    time.sleep(0.5)  # This will be detected
    threading.Thread(target=lambda: time.sleep(10)).start()  # This will NOT be detected

# Only detect thread leaks
@pytest.mark.no_leaks(tasks=False, threads=True, blocking=False)
def test_thread_only():
    threading.Thread(target=lambda: time.sleep(10)).start()  # This will be detected

Custom Configuration

import re

# Custom task name filtering
@pytest.mark.no_leaks(
    task_name_filter=re.compile(r"background-.*"),
    task_action="log"  # Log instead of raising
)
@pytest.mark.asyncio
async def test_custom_task_config():
    asyncio.create_task(asyncio.sleep(10), name="background-worker")  # Detected
    asyncio.create_task(asyncio.sleep(10), name="main-worker")  # Not detected

# Custom blocking threshold
@pytest.mark.no_leaks(
    blocking_threshold=0.5,  # Higher threshold
    blocking_action="warn"   # Warn instead of raising
)
@pytest.mark.asyncio
async def test_custom_blocking_config():
    time.sleep(0.3)  # Not detected (below threshold)
    time.sleep(0.6)  # Detected (above threshold)

Exception Handling in Tests

import pytest
from pyleak import TaskLeakError, ThreadLeakError, EventLoopBlockError, PyleakExceptionGroup

@pytest.mark.no_leaks
@pytest.mark.asyncio
async def test_leak_detection_failure():
    # This test will fail due to task leak
    asyncio.create_task(asyncio.sleep(10))

def test_manual_exception_handling():
    """Test that manually handles leak detection exceptions."""
    
    # Don't use the marker, handle exceptions manually
    from pyleak.combined import CombinedLeakDetector, PyLeakConfig
    from pyleak.utils import CallerContext
    
    config = PyLeakConfig()
    caller_context = CallerContext(filename=__file__, name="test_manual_exception_handling")
    
    try:
        with CombinedLeakDetector(config=config, is_async=False, caller_context=caller_context):
            threading.Thread(target=lambda: time.sleep(10)).start()
    except PyleakExceptionGroup as e:
        # Handle multiple leak types
        for error in e.exceptions:
            if isinstance(error, ThreadLeakError):
                print("Thread leak detected in manual test")
            # Handle other error types...

Real-World Test Examples

import pytest
import asyncio
import aiohttp
import requests
from pyleak import no_task_leaks, no_event_loop_blocking

class TestAsyncHTTPClient:
    
    @pytest.mark.no_leaks
    @pytest.mark.asyncio
    async def test_proper_async_http(self):
        """Test that properly uses async HTTP client."""
        async with aiohttp.ClientSession() as session:
            async with session.get('https://httpbin.org/get') as response:
                data = await response.json()
                assert response.status == 200
    
    @pytest.mark.no_leaks(blocking_threshold=0.1)
    @pytest.mark.asyncio 
    async def test_sync_http_blocking_detection(self):
        """This test will fail due to synchronous HTTP call blocking."""
        # This will be detected as blocking the event loop
        response = requests.get('https://httpbin.org/get')
        assert response.status_code == 200

class TestBackgroundTasks:
    
    @pytest.mark.no_leaks(enable_task_creation_tracking=True)
    @pytest.mark.asyncio
    async def test_background_task_cleanup(self):
        """Test proper cleanup of background tasks."""
        
        async def background_work():
            await asyncio.sleep(0.1)
            return "done"
        
        # Proper pattern - create and await task
        task = asyncio.create_task(background_work())
        result = await task
        assert result == "done"
        
        # Task is completed, no leak detected
    
    @pytest.mark.no_leaks
    @pytest.mark.asyncio
    async def test_leaked_task_detection(self):
        """This test will fail due to leaked background task."""
        
        async def long_running_task():
            await asyncio.sleep(10)
        
        # This task will be detected as leaked
        asyncio.create_task(long_running_task())
        await asyncio.sleep(0.1)

class TestThreadManagement:
    
    @pytest.mark.no_leaks
    def test_proper_thread_cleanup(self):
        """Test proper thread cleanup."""
        import threading
        
        def worker():
            time.sleep(0.1)
        
        thread = threading.Thread(target=worker)
        thread.start()
        thread.join()  # Proper cleanup
    
    @pytest.mark.no_leaks(grace_period=0.05)
    def test_thread_leak_detection(self):
        """This test will fail due to thread leak."""
        
        def long_running_worker():
            time.sleep(10)
        
        # Thread won't finish in time
        threading.Thread(target=long_running_worker).start()

Custom Test Fixtures

import pytest
from pyleak.combined import CombinedLeakDetector, PyLeakConfig
from pyleak.utils import CallerContext

@pytest.fixture
def leak_detector():
    """Fixture providing custom leak detection configuration."""
    config = PyLeakConfig()
    config.task_action = "log"  # Log instead of raising
    config.blocking_threshold = 0.05  # More sensitive
    return config

@pytest.mark.asyncio
async def test_with_custom_detector(leak_detector):
    """Test using custom leak detector configuration."""
    caller_context = CallerContext(filename=__file__, name="test_with_custom_detector")
    
    async with CombinedLeakDetector(
        config=leak_detector, 
        is_async=True, 
        caller_context=caller_context
    ):
        await asyncio.sleep(0.1)

Debugging Failed Tests

import pytest
import asyncio
from pyleak import TaskLeakError

def test_debug_task_leaks():
    """Example of debugging task leaks in tests."""
    
    try:
        # Manually use leak detection for debugging
        async def run_test():
            async with no_task_leaks(action="raise", enable_creation_tracking=True):
                # Test code that might leak tasks
                task1 = asyncio.create_task(asyncio.sleep(1), name="worker-1")
                task2 = asyncio.create_task(asyncio.sleep(2), name="worker-2")
                await asyncio.sleep(0.1)  # Not enough time for tasks to complete
        
        asyncio.run(run_test())
        
    except TaskLeakError as e:
        # Debug information
        print(f"Test failed: {e.task_count} leaked tasks")
        for task_info in e.leaked_tasks:
            print(f"Leaked task: {task_info.name}")
            if task_info.creation_stack:
                print("Created at:")
                print(task_info.format_creation_stack())
        
        # Re-raise to fail the test
        raise

Install with Tessl CLI