Best Python code snippet using avocado_python
runner.py
Source:runner.py
...154 if paused_msg:155 self.job.log.warning(paused_msg)156 else: # test_status157 self.status = msg158 def _add_status_failures(self, test_state):159 """160 Append TestStatus error to test_state in case there were any.161 """162 if self._failed:163 return add_runner_failure(test_state, "ERROR", "TestStatus failed,"164 " see overall job.log for details.")165 return test_state166 def finish(self, proc, started, step, deadline, result_dispatcher):167 """168 Wait for the test process to finish and report status or error status169 if unable to obtain the status till deadline.170 :param proc: The test's process171 :param started: Time when the test started172 :param first: Delay before first check173 :param step: Step between checks for the status174 :param deadline: Test execution deadline175 :param result_dispatcher: Result dispatcher (for test_progress176 notifications)177 """178 # Wait for either process termination or test status179 wait.wait_for(lambda: not proc.is_alive() or self.status, 1, 0, step)180 config = settings.as_dict()181 if self.status: # status exists, wait for process to finish182 timeout_process_alive = config.get('runner.timeout.process_alive')183 deadline = min(deadline, time.time() + timeout_process_alive)184 while time.time() < deadline:185 result_dispatcher.map_method('test_progress', False)186 if wait.wait_for(lambda: not proc.is_alive(), 1, 0, step):187 return self._add_status_failures(self.status)188 err = "Test reported status but did not finish"189 else: # proc finished, wait for late status delivery190 timeout_process_died = config.get('runner.timeout.process_died')191 deadline = min(deadline, time.time() + timeout_process_died)192 while time.time() < deadline:193 result_dispatcher.map_method('test_progress', False)194 if wait.wait_for(lambda: self.status, 1, 0, step):195 # Status delivered after the test process finished, pass196 return self._add_status_failures(self.status)197 err = "Test died without reporting the status."198 # At this point there were failures, fill the new test status199 TEST_LOG.debug("Original status: %s", str(self.status))200 test_state = self.early_status201 test_state['time_start'] = started202 test_state['time_elapsed'] = time.time() - started203 test_state['fail_reason'] = err204 test_state['status'] = exceptions.TestAbortError.status205 test_state['fail_class'] = (exceptions.TestAbortError.__class__.206 __name__)207 test_state['traceback'] = 'Traceback not available'208 try:209 with open(test_state['logfile'], 'r') as log_file_obj:210 test_state['text_output'] = log_file_obj.read()211 except IOError:212 test_state["text_output"] = "Not available, file not created yet"213 TEST_LOG.error('ERROR %s -> TestAbortError: %s.', err,214 test_state['name'])215 if proc.is_alive():216 TEST_LOG.warning("Killing hanged test process %s", proc.pid)217 os.kill(proc.pid, signal.SIGTERM)218 if not wait.wait_for(lambda: not proc.is_alive(), 1, 0, 0.01):219 os.kill(proc.pid, signal.SIGKILL)220 end_time = time.time() + 60221 while time.time() < end_time:222 if not proc.is_alive():223 break224 time.sleep(0.1)225 else:226 raise exceptions.TestError("Unable to destroy test's "227 "process (%s)" % proc.pid)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!