0
# PyTest Integration
1

2
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.
3

4
## Capabilities
5

6
### Pytest Plugin Registration
7

8
The plugin is automatically registered when pyleak is installed.
9

10
```python { .api }
11
# Entry point configuration (in pyproject.toml):
12
[project.entry-points."pytest11"]
13
pyleak = "pyleak.pytest_plugin"
14
```
15

16
### Test Marker
17

18
Primary marker for enabling leak detection in tests.
19

20
```python { .api }
21
@pytest.mark.no_leaks  # Enable all leak detection types
22
@pytest.mark.no_leaks(tasks=True, threads=False, blocking=True)  # Selective detection
23
def test_function(): ...
24

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

28
@pytest.mark.no_leaks("all")  # Enable all types (equivalent to no arguments)
29
def test_function(): ...
30
```
31

32
### Configuration Classes
33

34
Configuration class for customizing leak detection behavior.
35

36
```python { .api }
37
class PyLeakConfig:
38
    """Configuration for pyleak detection."""
39
    
40
    # Task detection settings
41
    tasks: bool = True
42
    task_action: str = "raise"
43
    task_name_filter: str | None = None
44
    enable_task_creation_tracking: bool = False
45
    
46
    # Thread detection settings
47
    threads: bool = True
48
    thread_action: str = "raise"
49
    thread_name_filter: str | None = DEFAULT_THREAD_NAME_FILTER
50
    exclude_daemon_threads: bool = True
51
    
52
    # Event loop blocking detection settings
53
    blocking: bool = True  
54
    blocking_action: str = "raise"
55
    blocking_threshold: float = 0.2
56
    blocking_check_interval: float = 0.01
57
    
58
    @classmethod
59
    def from_marker_args(cls, marker_args: dict[str, Any]) -> "PyLeakConfig":
60
        """Create configuration from pytest marker arguments."""
61
    
62
    def to_markdown_table(self) -> str:
63
        """Generate markdown table of configuration options."""
64
```
65

66
Combined leak detector for coordinating multiple detection types.
67

68
```python { .api }
69
class CombinedLeakDetector:
70
    """Combined detector for all leak types."""
71
    
72
    def __init__(
73
        self,
74
        config: PyLeakConfig,
75
        is_async: bool,
76
        caller_context: CallerContext | None = None,
77
    ): ...
78
    
79
    async def __aenter__(self): ...
80
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
81
    def __enter__(self): ...
82
    def __exit__(self, exc_type, exc_val, exc_tb): ...
83
```
84

85
### Plugin Functions
86

87
Function to determine if a test should be monitored.
88

89
```python { .api }
90
def should_monitor_test(item: pytest.Function) -> PyLeakConfig | None:
91
    """
92
    Check if test should be monitored and return config.
93
    
94
    Args:
95
        item: pytest Function item
96
        
97
    Returns:
98
        PyLeakConfig if test should be monitored, None otherwise
99
    """
100
```
101

102
Pytest hook for wrapping test execution.
103

104
```python { .api }
105
@pytest.hookimpl(hookwrapper=True)
106
def pytest_runtest_call(item: pytest.Function):
107
    """Wrap test execution with leak detection."""
108
```
109

110
## Configuration Options
111

112
### Marker Arguments
113

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

116
| Name | Default | Description |
117
|:------|:------|:------|
118
| tasks | True | Whether to detect task leaks |
119
| task_action | raise | Action to take when a task leak is detected |
120
| task_name_filter | None | Filter to apply to task names |
121
| enable_task_creation_tracking | False | Whether to enable task creation tracking |
122
| threads | True | Whether to detect thread leaks |
123
| thread_action | raise | Action to take when a thread leak is detected |
124
| thread_name_filter | DEFAULT_THREAD_NAME_FILTER | Filter to apply to thread names (default: exclude asyncio threads) |
125
| exclude_daemon_threads | True | Whether to exclude daemon threads |
126
| blocking | True | Whether to detect event loop blocking |
127
| blocking_action | raise | Action to take when a blocking event loop is detected |
128
| blocking_threshold | 0.2 | Threshold for blocking event loop detection |
129
| blocking_check_interval | 0.01 | Interval for checking for blocking event loop |
130
131
## Usage Examples
132

133
### Basic Setup
134

135
Add the marker to your pytest configuration:
136

137
**pyproject.toml**
138
```toml
139
[tool.pytest.ini_options]
140
markers = [
141
    "no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking"
142
]
143
```
144

145
**pytest.ini**
146
```ini
147
[tool:pytest]
148
markers = no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking
149
```
150

151
**conftest.py**
152
```python
153
import pytest
154

155
def pytest_configure(config):
156
    config.addinivalue_line(
157
        "markers", 
158
        "no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking"
159
    )
160
```
161

162
### Basic Test Detection
163

164
```python
165
import pytest
166
import asyncio
167
import threading
168
import time
169

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

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

182
### Selective Detection
183

184
```python
185
# Only detect task leaks and event loop blocking
186
@pytest.mark.no_leaks(tasks=True, blocking=True, threads=False)
187
@pytest.mark.asyncio
188
async def test_selective_detection():
189
    asyncio.create_task(asyncio.sleep(10))  # This will be detected
190
    time.sleep(0.5)  # This will be detected
191
    threading.Thread(target=lambda: time.sleep(10)).start()  # This will NOT be detected
192

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

199
### Custom Configuration
200

201
```python
202
import re
203

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

214
# Custom blocking threshold
215
@pytest.mark.no_leaks(
216
    blocking_threshold=0.5,  # Higher threshold
217
    blocking_action="warn"   # Warn instead of raising
218
)
219
@pytest.mark.asyncio
220
async def test_custom_blocking_config():
221
    time.sleep(0.3)  # Not detected (below threshold)
222
    time.sleep(0.6)  # Detected (above threshold)
223
```
224

225
### Exception Handling in Tests
226

227
```python
228
import pytest
229
from pyleak import TaskLeakError, ThreadLeakError, EventLoopBlockError, PyleakExceptionGroup
230

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

237
def test_manual_exception_handling():
238
    """Test that manually handles leak detection exceptions."""
239
    
240
    # Don't use the marker, handle exceptions manually
241
    from pyleak.combined import CombinedLeakDetector, PyLeakConfig
242
    from pyleak.utils import CallerContext
243
    
244
    config = PyLeakConfig()
245
    caller_context = CallerContext(filename=__file__, name="test_manual_exception_handling")
246
    
247
    try:
248
        with CombinedLeakDetector(config=config, is_async=False, caller_context=caller_context):
249
            threading.Thread(target=lambda: time.sleep(10)).start()
250
    except PyleakExceptionGroup as e:
251
        # Handle multiple leak types
252
        for error in e.exceptions:
253
            if isinstance(error, ThreadLeakError):
254
                print("Thread leak detected in manual test")
255
            # Handle other error types...
256
```
257

258
### Real-World Test Examples
259

260
```python
261
import pytest
262
import asyncio
263
import aiohttp
264
import requests
265
from pyleak import no_task_leaks, no_event_loop_blocking
266

267
class TestAsyncHTTPClient:
268
    
269
    @pytest.mark.no_leaks
270
    @pytest.mark.asyncio
271
    async def test_proper_async_http(self):
272
        """Test that properly uses async HTTP client."""
273
        async with aiohttp.ClientSession() as session:
274
            async with session.get('https://httpbin.org/get') as response:
275
                data = await response.json()
276
                assert response.status == 200
277
    
278
    @pytest.mark.no_leaks(blocking_threshold=0.1)
279
    @pytest.mark.asyncio 
280
    async def test_sync_http_blocking_detection(self):
281
        """This test will fail due to synchronous HTTP call blocking."""
282
        # This will be detected as blocking the event loop
283
        response = requests.get('https://httpbin.org/get')
284
        assert response.status_code == 200
285

286
class TestBackgroundTasks:
287
    
288
    @pytest.mark.no_leaks(enable_task_creation_tracking=True)
289
    @pytest.mark.asyncio
290
    async def test_background_task_cleanup(self):
291
        """Test proper cleanup of background tasks."""
292
        
293
        async def background_work():
294
            await asyncio.sleep(0.1)
295
            return "done"
296
        
297
        # Proper pattern - create and await task
298
        task = asyncio.create_task(background_work())
299
        result = await task
300
        assert result == "done"
301
        
302
        # Task is completed, no leak detected
303
    
304
    @pytest.mark.no_leaks
305
    @pytest.mark.asyncio
306
    async def test_leaked_task_detection(self):
307
        """This test will fail due to leaked background task."""
308
        
309
        async def long_running_task():
310
            await asyncio.sleep(10)
311
        
312
        # This task will be detected as leaked
313
        asyncio.create_task(long_running_task())
314
        await asyncio.sleep(0.1)
315

316
class TestThreadManagement:
317
    
318
    @pytest.mark.no_leaks
319
    def test_proper_thread_cleanup(self):
320
        """Test proper thread cleanup."""
321
        import threading
322
        
323
        def worker():
324
            time.sleep(0.1)
325
        
326
        thread = threading.Thread(target=worker)
327
        thread.start()
328
        thread.join()  # Proper cleanup
329
    
330
    @pytest.mark.no_leaks(grace_period=0.05)
331
    def test_thread_leak_detection(self):
332
        """This test will fail due to thread leak."""
333
        
334
        def long_running_worker():
335
            time.sleep(10)
336
        
337
        # Thread won't finish in time
338
        threading.Thread(target=long_running_worker).start()
339
```
340

341
### Custom Test Fixtures
342

343
```python
344
import pytest
345
from pyleak.combined import CombinedLeakDetector, PyLeakConfig
346
from pyleak.utils import CallerContext
347

348
@pytest.fixture
349
def leak_detector():
350
    """Fixture providing custom leak detection configuration."""
351
    config = PyLeakConfig()
352
    config.task_action = "log"  # Log instead of raising
353
    config.blocking_threshold = 0.05  # More sensitive
354
    return config
355

356
@pytest.mark.asyncio
357
async def test_with_custom_detector(leak_detector):
358
    """Test using custom leak detector configuration."""
359
    caller_context = CallerContext(filename=__file__, name="test_with_custom_detector")
360
    
361
    async with CombinedLeakDetector(
362
        config=leak_detector, 
363
        is_async=True, 
364
        caller_context=caller_context
365
    ):
366
        await asyncio.sleep(0.1)
367
```
368

369
### Debugging Failed Tests
370

371
```python
372
import pytest
373
import asyncio
374
from pyleak import TaskLeakError
375

376
def test_debug_task_leaks():
377
    """Example of debugging task leaks in tests."""
378
    
379
    try:
380
        # Manually use leak detection for debugging
381
        async def run_test():
382
            async with no_task_leaks(action="raise", enable_creation_tracking=True):
383
                # Test code that might leak tasks
384
                task1 = asyncio.create_task(asyncio.sleep(1), name="worker-1")
385
                task2 = asyncio.create_task(asyncio.sleep(2), name="worker-2")
386
                await asyncio.sleep(0.1)  # Not enough time for tasks to complete
387
        
388
        asyncio.run(run_test())
389
        
390
    except TaskLeakError as e:
391
        # Debug information
392
        print(f"Test failed: {e.task_count} leaked tasks")
393
        for task_info in e.leaked_tasks:
394
            print(f"Leaked task: {task_info.name}")
395
            if task_info.creation_stack:
396
                print("Created at:")
397
                print(task_info.format_creation_stack())
398
        
399
        # Re-raise to fail the test
400
        raise
401
```