<commit_before>import multiprocessing as mp
import threading
from time import time

from coverage import coverage
from specter.spec import TestEvent, DescribeEvent


class ExecuteTestProcess(mp.Process):
    def __init__(self, work_queue, all_cases, all_parents,
                 pipe, track_coverage=False, coverage_omit=None):
        super(ExecuteTestProcess, self).__init__()
        self.work_queue = work_queue
        self.all_cases = all_cases
        self.all_parents = all_parents
        self.worked = mp.Value('i', 0)
        self.pipe = pipe
        self.coverage = None

        if track_coverage:
            self.coverage = coverage(omit=coverage_omit, data_suffix=True)
            self.coverage._warn_no_data = False

    def run(self):
        last_time = time()
        completed = []

        if self.coverage:
            self.coverage.start()

        while True:
            # Get item and get real function to execute
            case_wrapper = self.work_queue.get()
            if case_wrapper == 'STOP':
                # Make sure buffer is cleared
                if len(completed) > 0:
                    self.pipe.send(completed)
                    self.completed = []<commit_msg>Disabling coverage in multiprocess run<commit_after>import multiprocessing as mp
import threading
from time import time

from coverage import coverage
from specter.spec import TestEvent, DescribeEvent


class ExecuteTestProcess(mp.Process):
    def __init__(self, work_queue, all_cases, all_parents,
                 pipe, track_coverage=False, coverage_omit=None):
        super(ExecuteTestProcess, self).__init__()
        self.work_queue = work_queue
        self.all_cases = all_cases
        self.all_parents = all_parents
        self.worked = mp.Value('i', 0)
        self.pipe = pipe
        self.coverage = None

        if track_coverage:
            self.coverage = coverage(omit=coverage_omit, data_suffix=True)
            self.coverage._warn_no_data = False

    def run(self):  # pragma: no cover
        """ Note: CI Coverage is turned off due to it not showing covered
        even with there being tests that run this function.
        """
        last_time = time()
        completed = []

        if self.coverage:
            self.coverage.start()

        while True:
            # Get item and get real function to execute
            case_wrapper = self.work_queue.get()
            if case_wrapper == 'STOP':
                # Make sure buffer is cleared
                if len(completed) > 0:
                    self.pipe.send(completed)
                    self.completed = []
{'fraction_non_alphanumeric': 0.07642338555598013, 'fraction_numerical': 0.0015284677111196026, 'mean_word_length': 2.0945626477541373, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '309644', 'n_tokens_mistral': 732, 'n_tokens_neox': 694, 'n_words': 196}