Coverage incorrectly reports parallel processing function as missing

[ztr-risingtide]

Summary

When generating single-file coverage reports with versions 7.x.x of pytest-cov, methods that are certainly executed in parallel processing are reported as missing.

Expected vs actual result

See Reproducer code below.

Expected result: 100% coverage, no lines missing
Actual result:

> pytest ./tests/test_demo.py -v -x --cov=demo --cov-report=term-missing
=================================================================================================== test session starts ===================================================================================================
platform win32 -- Python 3.12.2, pytest-9.0.3, pluggy-1.6.0 -- C:\Users\ZTR\.venv\Scripts\python.exe
cachedir: .pytest_cache
rootdir: C:\Users\ZTR
configfile: pyproject.toml
plugins: anyio-4.12.1, asyncio-1.4.0, cov-7.1.0, env-1.6.0, order-1.4.0, xdist-3.8.0
asyncio: mode=Mode.AUTO, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 1 item                                                                                                                                                                                                           

tests/test_demo.py::test_make_request_and_retrieve_solution PASSED                                                                                                                                                   [100%]

===================================================================================================== tests coverage ======================================================================================================
_____________________________________________________________________________________ coverage: platform win32, python 3.12.2-final-0 _____________________________________________________________________________________

Name                      Stmts   Miss  Cover   Missing
-------------------------------------------------------
src\demo.py                  42      9    79%   56-74
-------------------------------------------------------
TOTAL                        42      9    79%
==================================================================================================== 1 passed in 0.76s ====================================================================================================

Reproducer

Versions

Python 3.12.2
pytest 9.0.3
pytest-cov>6.3.0 (pytest-cov 6.1.1 through 6.3.0 work as expected)
aiomultiprocess==0.9.1

Config

pyproject.toml

[tool.pytest.ini_options]
asyncio_mode = "auto"

[tool.coverage.run]
concurrency = ["greenlet", "thread"]  # purposely omit "multiprocessing" because it causes coverage to dump more files than we care about
data_file = ".coverage/coverage"

[tool.coverage.report]
show_missing = true
skip_covered = false
skip_empty = true

exclude_lines = [
  "pragma: no cover",
  "raise AssertionError",
  "raise NotImplementedError",
  "if __name__ == .__main__.:",
  "if TYPE_CHECKING:",
  "if typing.TYPE_CHECKING:",
]

[tool.coverage.html]
directory = ".coverage/htmlcov"

[tool.coverage.xml]
output = ".coverage/coverage.xml"

[tool.coverage.lcov]
output = ".coverage/lcov.info"

Code

src\demo.py

# -*- coding: utf-8 -*-

# %% Imports

import time as timer
from typing import Optional

from aiomultiprocess import Pool


# %% Request class


class DemoRequest:
    def __init__(
        self,
    ) -> None:
        self._solve_tasks: dict[str, tuple] = {}
        self._solutions: dict[str, str | None] = {}
        self._counter: int = 0
        self._solve_called: bool = True

    def _reset(self):
        self._solve_tasks = {}
        self._solutions = {}
        self._solve_called = False

    async def make_request(
        self,
        param_a: int,
        param_b: str,
    ) -> tuple[Optional[str], dict]:
        if self._solve_called:
            # Reset so that we do not try to process previously completed solve tasks again
            # if new requests are being made to this instance
            self._reset()

        id_str = str(self._counter)
        self._counter += 1

        request_dict: dict = {
            "param_a": param_a,
            "param_b": param_b,
        }

        # Store up argument tuples as a collection of "tasks" to pass to solve() - we will
        # eventually run all of these tasks in parallel
        self._solve_tasks[id_str] = (param_a, param_b)

        return id_str, request_dict

    async def _solve_with_timer(
        self,
        request_tuple: tuple[str, tuple],
    ) -> tuple[str, str | None]:
        id_str, args = request_tuple
        (param_a, param_b) = args

        print(f"Solving task {id_str}...")
        t0 = timer.time()

        try:
            solution = f"{id_str} ({param_a}, {param_b}) solved!"

        except Exception:  # pragma: no cover
            # Catching exception prevents the entire parallel pool from crashing if an
            # early task fails
            print(f"Error solving task {id_str}:")
            return id_str, None

        t_delta = timer.time() - t0
        print(f"Solving task {id_str} took {t_delta:.4f} s")

        return id_str, solution

    async def retrieve_solution(
        self,
        id_str: str,
    ) -> str | None:
        result_list = []
        async with Pool(
            processes=3,  # Maximum number of worker processes
            childconcurrency=1,  # Allow only one queue of tasks from which all workers must draw
            queuecount=1,  # Allow each worker to run only one task at a time
        ) as pool:
            async for solution_tuple in pool.map(
                self._solve_with_timer, list(self._solve_tasks.items())
            ):
                # Cannot save results directly to dict self._solutions because changes to the
                # dict are not synchronised between the processes - this can cause
                # "RuntimeError: dictionary changed size during iteration"
                result_list.append(solution_tuple)

            pool.close()
            await pool.join()

        for solution_tuple in result_list:
            solution_id, solution = solution_tuple
            self._solutions[solution_id] = solution

        return self._solutions[id_str]


# %%

tests\test_demo.py

# -*- coding: utf-8 -*-


# %% Imports

import pytest

from demo import DemoRequest


# %% Fixtures


@pytest.fixture
def my_request() -> DemoRequest:
    my_request = DemoRequest()
    return my_request


# %% Tests


@pytest.mark.asyncio
async def test_make_request_and_retrieve_solution(
    my_request: DemoRequest,
) -> None:
    # --- Arrange ---
    param_a = 2
    param_b = "b"

    # --- Act ---
    id_str, request_dict = await my_request.make_request(
        param_a=param_a,
        param_b=param_b,
    )

    # --- Assert ---
    assert id_str is not None

    # --- Act ---
    solution_str = await my_request.retrieve_solution(id_str)

    # --- Assert ---
    assert solution_str is not None
    assert str(param_a) in solution_str
    assert param_b in solution_str