nested functions in multiprocessing not considered covered

[Ryang20718]

Describe the bug If I have a nested function that I call with multiprocessing, it is deemed not covered

To Reproduce How can we reproduce the problem? Please be specific. Don't link to a failing CI job. Answer the questions below:

Python 3.10.7 coverage.py 7.6, pytest-cov 5.0.0, pytest 8.2.2

implementation.py

import multiprocessing


def _sample_linestring(annotation, sample_dist, level):
    # This line often doesn't show up in coverage without proper config
    return f"Processed {annotation} at level {level} with dist {sample_dist}"


def sample_annotations(annotations, sample_dist):
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        return pool.starmap(_sample_linestring, [(annotation, sample_dist, level) for annotation, level in annotations])


def the_answer():
    """Example function to be tested."""
    annotations = [(f"annotation_{i}", i) for i in range(5)]
    sample_dist = 10
    result = sample_annotations(annotations, sample_dist)

test_implementation.py

from implementation import the_answer

def test_the_answer():
    answer = the_answer()
    assert answer == 42


test_the_answer()

.coveragerc

[run]
parallel = True
concurrency = multiprocessing,thread

1. pip install pytest==8.2.2 pytest-cov=5.0.0 coverage==7.6
2. pytest --cov=implementation implementation_test.py --cov-report=term-missing

Expected behavior We see missing coverage

def _sample_linestring(annotation, sample_dist, level):
    # This line often doesn't show up in coverage without proper config
    return f"Processed {annotation} at level {level} with dist {sample_dist}" # <-- missing coverage

Additional context Add any other context about the problem here.

[sebsken]

According to the doc, it is a best practice to close and join the pool: https://pytest-cov.readthedocs.io/en/latest/subprocess-support.html#if-you-use-multiprocessing Did you try that?

with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
    a = pool.starmap(_sample_linestring, [(annotation, sample_dist, level) for annotation, level in annotations])
    pool.close()
    pool.join()
    return a