|
| 1 | +# In conftest.py or test_logging_utils.py |
| 2 | +import logging |
| 3 | +from collections.abc import AsyncIterator |
| 4 | +from contextlib import asynccontextmanager |
| 5 | +from unittest.mock import patch |
| 6 | + |
| 7 | +import pytest |
| 8 | +from servicelib.logging_utils import setup_async_loggers_lifespan |
| 9 | + |
| 10 | + |
| 11 | +@pytest.fixture(autouse=True) |
| 12 | +def preserve_caplog_for_async_logging(request): |
| 13 | + """Automatically preserve caplog handlers when both caplog and async logging are used.""" |
| 14 | + # Check if this test uses caplog fixture |
| 15 | + if "caplog" not in request.fixturenames: |
| 16 | + yield # No caplog, no patching needed |
| 17 | + return |
| 18 | + |
| 19 | + # Patch setup_async_loggers_lifespan to preserve caplog handlers |
| 20 | + original_setup = setup_async_loggers_lifespan |
| 21 | + |
| 22 | + @asynccontextmanager |
| 23 | + async def patched_setup_async_loggers_lifespan(**kwargs) -> AsyncIterator[None]: |
| 24 | + # Find caplog's handler in root logger |
| 25 | + root_logger = logging.getLogger() |
| 26 | + caplog_handlers = [ |
| 27 | + h for h in root_logger.handlers if "LogCaptureHandler" in f"{type(h)}" |
| 28 | + ] |
| 29 | + |
| 30 | + async with original_setup(**kwargs): |
| 31 | + # After setup, restore caplog handlers alongside queue handler |
| 32 | + for handler in caplog_handlers: |
| 33 | + if handler not in root_logger.handlers: |
| 34 | + root_logger.addHandler(handler) |
| 35 | + yield |
| 36 | + |
| 37 | + with ( |
| 38 | + patch( |
| 39 | + "tests.test_logging_utils.setup_async_loggers_lifespan", |
| 40 | + patched_setup_async_loggers_lifespan, |
| 41 | + ), |
| 42 | + patch( |
| 43 | + "servicelib.logging_utils.setup_async_loggers_lifespan", |
| 44 | + patched_setup_async_loggers_lifespan, |
| 45 | + ), |
| 46 | + ): |
| 47 | + yield |
0 commit comments