Enabling Python 3.5 Coverage support for multiprocessing Pool

mp_coverage.py

import logging
import os

from multiprocessing.pool import Pool as OrigPool
from coverage import Coverage
from multiprocessing.context import Process, DefaultContext, _concrete_contexts

logger = logging.getLogger(__name__)


class CoverageProcess(Process):
    def run(self):
        should_cov = os.environ.get('COVERAGE_PROCESS_START', False)

        if should_cov:
            logger.debug('Starting coverage')
            cov = Coverage(data_suffix=True, config_file='.coveragerc')
            cov._warn_no_data = True
            cov._warn_unimported_source = True
            cov.start()

            try:
                super().run()
            finally:
                logger.debug('Stopping coverage')
                cov.stop()
                cov.save()
        else:
            super().run()


class CoverageContext(DefaultContext):
    Process = CoverageProcess


def Pool(*args, **kwargs):
    ctx = CoverageContext(_concrete_contexts['fork'])
    return OrigPool(*args, **kwargs, context=ctx)

Hi, I believe this link might help you: http://pytest-cov.readthedocs.io/en/latest/mp.html#ungraceful-process-shutdown

I have managed to track coverage within a Pool simply by adding calls to close and join, as follows:

with Pool(min(cpu_count(), len(tasks))) as p:
       list(p.imap(worker, tasks))
       p.close()
       p.join()