Method check_case

tests_python/debugger_unittest.py:446–497

(self, writer_class, wait_for_port=True, wait_for_initialization=True, dap=False)

    @contextmanager
    def check_case(self, writer_class, wait_for_port=True, wait_for_initialization=True, dap=False):
        try:
            if callable(writer_class):
                writer = writer_class()
            else:
                writer = writer_class
            try:
                writer.start()
                if wait_for_port:
                    wait_for_condition(lambda: hasattr(writer, "port"))
                self.writer = writer

                args = self.get_command_line()

                args = self.add_command_line_args(args, dap=dap)

                if SHOW_OTHER_DEBUG_INFO:
                    print("executing: %s" % (" ".join(args),))

                with self.run_process(args, writer) as dct_with_stdout_stder:
                    try:
                        if not wait_for_initialization:
                            # The use-case for this is that the debugger can't even start-up in this
                            # scenario, as such, sleep a bit so that the output can be collected.
                            time.sleep(1)
                        elif wait_for_port:
                            wait_for_condition(lambda: writer.finished_initialization)
                    except TimeoutError:
                        sys.stderr.write("Timed out waiting for initialization\n")
                        sys.stderr.write(
                            "stdout:\n%s\n\nstderr:\n%s\n"
                            % (
                                "".join(dct_with_stdout_stder["stdout"]),
                                "".join(dct_with_stdout_stder["stderr"]),
                            )
                        )
                        raise
                    finally:
                        writer.get_stdout = lambda: "".join(dct_with_stdout_stder["stdout"])
                        writer.get_stderr = lambda: "".join(dct_with_stdout_stder["stderr"])

                    yield writer
            finally:
                writer.do_kill()
                writer.log = []

            stdout = dct_with_stdout_stder["stdout"]
            stderr = dct_with_stdout_stder["stderr"]
            writer.additional_output_checks("".join(stdout), "".join(stderr))
        except IgnoreFailureError:
            sys.stderr.write("Test finished with ignored failure.\n")
            return

    def create_process(self, args, writer):
        env = writer.get_environ() if writer is not None else None