Thread Leak Detection

def no_thread_leaks(
    action: str = "warn",
    name_filter: Optional[Union[str, re.Pattern]] = DEFAULT_THREAD_NAME_FILTER,
    logger: Optional[logging.Logger] = None,
    exclude_daemon: bool = True,
    grace_period: float = 0.1,
):
    """
    Context manager/decorator that detects thread leaks within its scope.

    Args:
        action: Action to take when leaks are detected ("warn", "log", "cancel", "raise")
        name_filter: Optional filter for thread names (string or regex)
        logger: Optional logger instance
        exclude_daemon: Whether to exclude daemon threads from detection
        grace_period: Time to wait for threads to finish naturally (seconds)

    Returns:
        _ThreadLeakContextManager: Context manager that can also be used as decorator

    Example:
        # As context manager
        with no_thread_leaks():
            threading.Thread(target=some_function).start()

        # As decorator
        @no_thread_leaks(action="raise")
        def my_function():
            threading.Thread(target=some_work).start()
    """

class ThreadLeakError(LeakError):
    """Raised when thread leaks are detected and action is set to RAISE."""

DEFAULT_THREAD_NAME_FILTER: re.Pattern
# Compiled regex pattern: re.compile(r"^(?!asyncio_\d+$).*")
# Excludes asyncio internal threads (names matching "asyncio_\d+")

import threading
import time
from pyleak import no_thread_leaks

def worker_function():
    time.sleep(5)

def main():
    with no_thread_leaks():
        # This thread will be detected as leaked if it's still running
        threading.Thread(target=worker_function).start()
        time.sleep(0.1)  # Brief pause, not enough for thread to complete

import threading
from pyleak import ThreadLeakError, no_thread_leaks

def long_running_worker():
    time.sleep(10)

def main():
    try:
        with no_thread_leaks(action="raise"):
            thread = threading.Thread(target=long_running_worker)
            thread.start()
            # Thread is still running when context exits
    except ThreadLeakError as e:
        print(f"Thread leak detected: {e}")

import re
from pyleak import no_thread_leaks

# Filter by exact name
with no_thread_leaks(name_filter="worker-thread"):
    thread = threading.Thread(target=some_work, name="worker-thread")
    thread.start()

# Filter by regex pattern
with no_thread_leaks(name_filter=re.compile(r"background-.*")):
    thread = threading.Thread(target=some_work, name="background-processor")
    thread.start()

# Use default filter (excludes asyncio threads)
with no_thread_leaks():  # DEFAULT_THREAD_NAME_FILTER is used
    pass

# Warn mode (default) - issues ResourceWarning
with no_thread_leaks(action="warn"):
    pass

# Log mode - writes to logger
with no_thread_leaks(action="log"):
    pass

# Cancel mode - warns that threads can't be force-stopped
with no_thread_leaks(action="cancel"):
    pass  # Will warn about inability to force-stop threads

# Raise mode - raises ThreadLeakError
with no_thread_leaks(action="raise"):
    pass

import threading
from pyleak import no_thread_leaks

def daemon_worker():
    time.sleep(10)

def regular_worker():
    time.sleep(10)

# Exclude daemon threads (default behavior)
with no_thread_leaks(exclude_daemon=True):
    # This daemon thread will be ignored
    daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
    daemon_thread.start()
    
    # This regular thread will be detected if leaked
    regular_thread = threading.Thread(target=regular_worker)
    regular_thread.start()

# Include daemon threads in detection
with no_thread_leaks(exclude_daemon=False):
    # Both daemon and regular threads will be monitored
    daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
    daemon_thread.start()

from pyleak import no_thread_leaks

def quick_worker():
    time.sleep(0.05)  # Very quick work

# Short grace period
with no_thread_leaks(grace_period=0.01):
    threading.Thread(target=quick_worker).start()
    # May detect thread as leaked due to short grace period

# Longer grace period
with no_thread_leaks(grace_period=0.2):
    threading.Thread(target=quick_worker).start()
    # Thread has time to complete naturally

@no_thread_leaks(action="raise")
def my_threaded_function():
    # Any leaked threads will cause ThreadLeakError to be raised
    thread = threading.Thread(target=some_background_work)
    thread.start()
    thread.join()  # Proper cleanup

import threading
from pyleak import no_thread_leaks

def background_task():
    time.sleep(2)

# Proper cleanup pattern
def main():
    with no_thread_leaks(action="raise"):
        thread = threading.Thread(target=background_task)
        thread.start()
        thread.join()  # Wait for thread to complete
        # No leak detected because thread is properly joined

import pytest
import threading
from pyleak import no_thread_leaks, ThreadLeakError

def test_thread_cleanup():
    """Test that ensures threads are properly cleaned up."""
    def worker():
        time.sleep(0.5)
    
    with pytest.raises(ThreadLeakError):
        with no_thread_leaks(action="raise", grace_period=0.1):
            # This thread won't finish in time
            threading.Thread(target=worker).start()

def test_proper_thread_management():
    """Test that properly managed threads don't trigger leaks."""
    def worker():
        time.sleep(0.1)
    
    # This should not raise an exception
    with no_thread_leaks(action="raise", grace_period=0.2):
        thread = threading.Thread(target=worker)
        thread.start()
        thread.join()

import re
from pyleak import no_thread_leaks

# Only monitor specific thread patterns
custom_filter = re.compile(r"^(worker|processor|handler)-.*")

with no_thread_leaks(name_filter=custom_filter, action="raise"):
    # These threads will be monitored
    threading.Thread(target=work, name="worker-1").start()
    threading.Thread(target=process, name="processor-main").start()
    
    # This thread will be ignored
    threading.Thread(target=utility, name="utility-helper").start()