Method run

Lib/test/libregrtest/run_workers.py:390–418

(self)

        return MultiprocessResult(result, stdout)

    def run(self) -> None:
        fail_fast = self.runtests.fail_fast
        fail_env_changed = self.runtests.fail_env_changed
        try:
            while not self._stopped:
                try:
                    test_name = next(self.pending)
                except StopIteration:
                    break

                self.start_time = time.monotonic()
                self.test_name = test_name
                try:
                    mp_result = self._runtest(test_name)
                except WorkerError as exc:
                    mp_result = exc.mp_result
                finally:
                    self.test_name = _NOT_RUNNING
                mp_result.result.duration = time.monotonic() - self.start_time
                self.output.put((False, mp_result))

                if mp_result.result.must_stop(fail_fast, fail_env_changed):
                    break
        except ExitThread:
            pass
        except BaseException:
            self.output.put((True, traceback.format_exc()))
        finally:
            self.output.put(WorkerThreadExited())

    def _wait_completed(self) -> None:
        popen = self._popen