为何pytest搭配loguru（enqueue=True时）无法识别AssertionError？

问题场景 #

我在conftest.py中参考Loguru官方文档重定义了caplog fixture，由于业务代码涉及多进程，所以设置了enqueue=True：

@pytest.fixture
def caplog(caplog: LogCaptureFixture):
    """Redefine caplog to work with loguru, taken from their docs
    """
    handler_id = logger.add(
        caplog.handler,
        format="{message}",
        level=0,
        filter=lambda record: record["level"].no >= caplog.handler.level,
        enqueue=True,  # Set to 'True' if your test is spawning child processes.
    )
    yield caplog
    logger.remove(handler_id)

测试用例与异常现象 #

我写了一个基础测试用例，当enqueue=False时测试正常通过，但设置为True后出现异常：

• 将assert_no_errors_logged函数的代码直接复制到测试用例内部执行时，能正常触发AssertionError，符合预期；
• 但直接调用assert_no_errors_logged函数时，pytest报错提示未捕获到AssertionError。

测试代码如下：

def test_errors_are_caught(caplog):
    logger.debug("debug message")
    logger.info("an info")
    logger.error("an error")
    logger.critical("a very bad thing")
    with pytest.raises(AssertionError):
        assert_no_errors_logged(caplog)

def assert_no_errors_logged(caplog):
    error_messages = [record for record in caplog.records if record.levelno >= logging.ERROR]
    num_errors = len(error_messages)
    assert num_errors == 0

报错详情 #

============================== 1 failed in 0.28s ==============================
FAILED                [100%]2024-07-15 14:38:33.327 | DEBUG    | test_compute_statistics:test_errors_are_caught:440 - debug message
2024-07-15 14:38:33.327 | INFO     | test_compute_statistics:test_errors_are_caught:441 - an info
2024-07-15 14:38:33.327 | ERROR    | test_compute_statistics:test_errors_are_caught:442 - an error
2024-07-15 14:38:33.327 | CRITICAL | test_compute_statistics:test_errors_are_caught:443 - a very bad thing

tests\unit\test_compute_statistics.py:438 (test_errors_are_caught)
caplog = <_pytest.logging.LogCaptureFixture object at 0x000001A41CBB4550>

    def test_errors_are_caught(caplog):
        logger.debug("debug message")
        logger.info("an info")
        logger.error("an error")
        logger.critical("a very bad thing")
>       with pytest.raises(AssertionError):
E       Failed: DID NOT RAISE <class 'AssertionError'>

test_compute_statistics.py:444: Failed

Process finished with exit code 1

问题原因与解决方法 #

问题根源在于enqueue=True的异步特性：Loguru的队列handler会把日志消息放到后台队列异步处理，直接调用断言函数时，队列中的日志可能还未处理完成，导致caplog.records中没有错误日志，断言无法触发；而把代码复制到测试用例内部时，代码执行顺序或上下文的差异让队列日志已处理完毕，caplog.records能拿到完整日志。

有两种可行的解决方式：

1. 测试用例内等待队列清空
   在调用断言函数前，调用logger.complete()（Loguru 0.7.0+支持），该方法会阻塞直到队列中所有日志处理完成：

def test_errors_are_caught(caplog):
    logger.debug("debug message")
    logger.info("an info")
    logger.error("an error")
    logger.critical("a very bad thing")
    logger.complete()  # 等待队列日志处理完成
    with pytest.raises(AssertionError):
        assert_no_errors_logged(caplog)

2. 调整fixture清理逻辑
   在fixture的yield之后、移除handler之前，先调用logger.complete()，确保所有日志处理完毕再清理handler：

@pytest.fixture
def caplog(caplog: LogCaptureFixture):
    """Redefine caplog to work with loguru, taken from their docs
    """
    handler_id = logger.add(
        caplog.handler,
        format="{message}",
        level=0,
        filter=lambda record: record["level"].no >= caplog.handler.level,
        enqueue=True,  # Set to 'True' if your test is spawning child processes.
    )
    yield caplog
    logger.complete()  # 先处理完队列日志再移除handler
    logger.remove(handler_id)

两种方式都能保证caplog.records包含所有日志，断言可以正常触发。

内容的提问来源于stack exchange，提问作者codeananda