Method run_process

tests_python/debugger_unittest.py:519–650

(self, args, writer)

    @contextmanager
    def run_process(self, args, writer):
        process = self.create_process(args, writer)
        writer.process = process
        stdout = []
        stderr = []
        finish = [False]
        dct_with_stdout_stder = {}
        fail_with_message = False

        try:
            start_in_daemon_thread(read_process, (process.stdout, stdout, sys.stdout, "stdout", finish))
            start_in_daemon_thread(read_process, (process.stderr, stderr, sys.stderr, "stderr", finish))

            if SHOW_OTHER_DEBUG_INFO:
                print("Both processes started")

            # polls can fail (because the process may finish and the thread still not -- so, we give it some more chances to
            # finish successfully).
            initial_time = time.time()
            shown_intermediate = False
            dumped_threads = False

            dct_with_stdout_stder["stdout"] = stdout
            dct_with_stdout_stder["stderr"] = stderr
            try:
                yield dct_with_stdout_stder
            except:
                fail_with_message = True
                # Let's print the actuayl exception here (it doesn't appear properly on Python 2 and
                # on Python 3 it's hard to find because pytest output is too verbose).
                sys.stderr.write("***********\n")
                sys.stderr.write("***********\n")
                sys.stderr.write("***********\n")
                traceback.print_exc()
                sys.stderr.write("***********\n")
                sys.stderr.write("***********\n")
                sys.stderr.write("***********\n")
                raise

            if not writer.finished_ok:
                self.fail_with_message(
                    "The thread that was doing the tests didn't finish successfully (writer.finished_ok = True not set).",
                    stdout,
                    stderr,
                    writer,
                )

            while True:
                if process.poll() is not None:
                    if writer.EXPECTED_RETURNCODE != "any":
                        expected_returncode = writer.EXPECTED_RETURNCODE
                        if not isinstance(expected_returncode, (list, tuple)):
                            expected_returncode = (expected_returncode,)

                        if process.returncode not in expected_returncode:
                            self.fail_with_message(
                                "Expected process.returncode to be %s. Found: %s" % (writer.EXPECTED_RETURNCODE, process.returncode),
                                stdout,