How to use _process_started method in autotest

Best Python code snippet using autotest_python

postjob_task.py

Source:postjob_task.py Github

copy

Full Screen

...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')...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful