KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _process_started method in autotest

Best Python code snippet using autotest_python

postjob_task.py

Source:postjob_task.py

...105        raise NotImplementedError106    @classmethod107    def _can_run_new_process(cls):108        return cls._num_running_processes < cls._max_processes()109    def _process_started(self):110        return bool(self.monitor)111    def tick(self):112        # override tick to keep trying to start until the process count goes113        # down and we can, at which point we revert to default behavior114        if self._process_started():115            super(SelfThrottledPostJobTask, self).tick()116        else:117            self._try_starting_process()118    def run(self):119        # override run() to not actually run unless we can120        self._try_starting_process()121    @classmethod122    def _notify_process_limit_hit(cls):123        """Send an email to notify that process limit is hit."""124        if cls._notification_on:125            subject = '%s: hitting max process limit.' % cls.__name__126            message = ('Running processes/Max processes: %d/%d'127                       % (cls._num_running_processes, cls._max_processes()))128            email_manager.manager.enqueue_notify_email(subject, message)129            cls._notification_on = False130    @classmethod131    def _reset_notification_switch_if_necessary(cls):132        """Reset _notification_on if necessary.133        Set _notification_on to True on the following cases:134        1) If the limit of max processes configuration changes;135        2) If _notification_on is False and the number of running processes136           drops to lower than a level defined in REVIVE_NOTIFICATION_THRESHOLD.137        """138        if cls._last_known_max_processes != cls._max_processes():139            cls._notification_on = True140            cls._last_known_max_processes = cls._max_processes()141            return142        percentage = float(cls._num_running_processes) / cls._max_processes()143        if (not cls._notification_on and144            percentage < cls.REVIVE_NOTIFICATION_THRESHOLD):145            cls._notification_on = True146    def _try_starting_process(self):147        self._reset_notification_switch_if_necessary()148        if not self._can_run_new_process():149            self._notify_process_limit_hit()150            return151        # actually run the command152        super(SelfThrottledPostJobTask, self).run()153        if self._process_started():154            self._increment_running_processes()155    def finished(self, success):156        super(SelfThrottledPostJobTask, self).finished(success)157        if self._process_started():158            self._decrement_running_processes()159class GatherLogsTask(PostJobTask):160    """161    Task responsible for162    * gathering uncollected logs (if Autoserv crashed hard or was killed)163    * copying logs to the results repository164    * spawning CleanupTasks for hosts, if necessary165    * spawning a FinalReparseTask for the job166    * setting the final status of the host, directly or through a cleanup167    """168    def __init__(self, queue_entries, recover_run_monitor=None):169        self._job = queue_entries[0].job170        super(GatherLogsTask, self).__init__(171            queue_entries, log_file_name='.collect_crashinfo.log')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.