Capturing logs does not work with multiprocessing

[starhel]

A detailed description of the bug or problem

Capturing logs from spawned processes is not working - all calls to logging are missing in caplog.records and terminal output (live logs are printed correctly with log_cli=1).

pip list and system version

Package        Version
-------------- -------
exceptiongroup 1.2.0
iniconfig      2.0.0
packaging      24.0
pip            23.2.1
pluggy         1.4.0
pytest         8.1.1
setuptools     68.1.2
tomli          2.0.1
wheel          0.41.2

System version: Ubuntu 22.04.3 LTS

Minimal example

import logging
import multiprocessing

def f(i):
    logging.getLogger("some_logger").error(f"Found {i}")

def run_multiprocessing():
    with multiprocessing.Pool(2) as pool:
        pool.map(f, [1, 2])
    f(3)

def test_multiprocessing_missing(caplog):
    run_multiprocessing()
    assert len(caplog.records) == 3  # failed: 1 == 3

Workaround

It's possible to add a workaround to this case when you know which logger you need to hack.

def _iter_queue(queue):
    sentinel = None
    queue.put(sentinel)
    return iter(queue.get, sentinel)

@contextlib.contextmanager
def fix_caplog(name):
    queue = multiprocessing.Queue()
    logger = logging.getLogger(name)
    handler = logging.handlers.QueueHandler(queue)
    old_handlers = copy(logger.handlers)
    old_propagate = logger.propagate
    for old_handler in old_handlers:
        logger.removeHandler(old_handler)
    logger.addHandler(handler)
    logger.propagate = False
    yield
    logger.propagate = old_propagate
    logger.removeHandler(handler)
    for old_handler in old_handlers:
        logger.addHandler(old_handler)
    for record in _iter_queue(queue):
        logger.handle(record)

def test_multiprocessing_fix(caplog):
    with fix_caplog("some_logger"):
        run_multiprocessing()
    assert len(caplog.records) == 3  # ok

As this workaround works correctly in my tests, I'm aware that it may not working correctly in some scenarios:

• hierarchical loggers
• parallel running tests, like pytest-xdist (probably, never tested)
• ...

As this example is just simple reproduction, I encountered this issue while implementing tests for code based on luigi framework, where certain code segments are executed in child processes. I am curious whether it is feasible to support multiprocessing in caplog without resorting to hacks, and I am open to preparing a PR with your guidance.

[RonnyPfannschmidt]

Pytest is unaware of multiprocessing

You need to add a passover handler yourself

[nicoddemus]

I am curious whether it is feasible to support multiprocessing in caplog without resorting to hacks, and I am open to preparing a PR with your guidance.

Not sure if it would be possible to implement this in a general manner, as it would require some form of hooking into the subprocess spawning behavior to capture and transfer the captured logging output....

If multiprocessing had some form of way to enable this automatically, I could see we adding support for it to pytest, but a more complicated implementation is probably out of scope for the pytest core. Perhaps a plugin could be a good fit.

I'm closing this for now, but feel free to follow up with more questions.

[benjamin-kirkbride]

You need to add a passover handler yourself

Anyone have an example or a link to more info?

[benjamin-kirkbride]

found more info: https://github.com/pytest-dev/pytest/issues/3037