Function cleanup_processes

tests/process_helpers.py:15–29

Reap every process on the way out, whatever the block did.

(processes: list[BaseProcess])

@contextmanager
def cleanup_processes(processes: list[BaseProcess]) -> Generator[None]:
    """Reap every process on the way out, whatever the block did."""
    try:
        yield
    finally:
        for process in processes:
            # Terminating an unstarted process raises over the assertion that escaped before it started.
            if process.pid is not None:
                process.terminate()
                process.join(timeout=_REAP_DEADLINE)
                if process.is_alive():  # pragma: no cover  # SIGTERM only loses this race on a saturated runner
                    process.kill()
                    process.join(timeout=_REAP_DEADLINE)
            # Left open, the finalizer runs during whichever test the collector lands in, as an unraisable exception.
            process.close()


__all__ = ["cleanup_processes"]