| """ | |
| Test lldb-dap runInTerminal reverse request | |
| """ | |
| from contextlib import contextmanager | |
| from lldbsuite.test.decorators import * | |
| from lldbsuite.test.lldbtest import line_number | |
| import lldbdap_testcase | |
| import os | |
| import subprocess | |
| import json | |
| @contextmanager | |
| def fifo(*args, **kwargs): | |
| if sys.platform == "win32": | |
| import ctypes | |
| comm_file = r"\\.\pipe\lldb-dap-run-in-terminal-comm" | |
| PIPE_ACCESS_DUPLEX = 0x00000003 | |
| PIPE_TYPE_MESSAGE = 0x00000004 | |
| PIPE_READMODE_MESSAGE = 0x00000002 | |
| PIPE_WAIT = 0x00000000 | |
| PIPE_UNLIMITED_INSTANCES = 255 | |
| kernel32 = ctypes.windll.kernel32 | |
| pipe = kernel32.CreateNamedPipeW( | |
| comm_file, | |
| PIPE_ACCESS_DUPLEX, | |
| PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, | |
| PIPE_UNLIMITED_INSTANCES, | |
| 4096, | |
| 4096, | |
| 0, | |
| None, | |
| ) | |
| else: | |
| comm_file = os.path.join(kwargs["directory"], "comm-file") | |
| pipe = None | |
| os.mkfifo(comm_file) | |
| try: | |
| yield comm_file, pipe | |
| finally: | |
| if pipe is not None: | |
| kernel32.DisconnectNamedPipe(pipe) | |
| kernel32.CloseHandle(pipe) | |
| def read_pipe_message(pipe): | |
| import ctypes | |
| ERROR_MORE_DATA = 234 | |
| kernel32 = ctypes.windll.kernel32 | |
| buffer = b"" | |
| while True: | |
| chunk = ctypes.create_string_buffer(4096) | |
| bytes_read = ctypes.wintypes.DWORD() | |
| success = kernel32.ReadFile(pipe, chunk, 4096, ctypes.byref(bytes_read), None) | |
| buffer += chunk.raw[: bytes_read.value] | |
| if success: | |
| break | |
| if ctypes.GetLastError() != ERROR_MORE_DATA: | |
| break | |
| return buffer.decode() | |
| @skipIfBuildType(["debug"]) | |
| class TestDAP_runInTerminal(lldbdap_testcase.DAPTestCaseBase): | |
| SHARED_BUILD_TESTCASE = False | |
| def read_pid_message(self, fifo_file, pipe): | |
| if sys.platform == "win32": | |
| import ctypes | |
| ctypes.windll.kernel32.ConnectNamedPipe(pipe, None) | |
| self.assertIn("pid", read_pipe_message(pipe)) | |
| else: | |
| with open(fifo_file, "r") as file: | |
| self.assertIn("pid", file.readline()) | |
| @staticmethod | |
| def send_did_attach_message(fifo_file, pipe=None): | |
| message = json.dumps({"kind": "didAttach"}) + "\n" | |
| if sys.platform == "win32": | |
| import ctypes | |
| kernel32 = ctypes.windll.kernel32 | |
| bytes_written = ctypes.wintypes.DWORD() | |
| kernel32.ConnectNamedPipe(pipe, None) | |
| kernel32.WriteFile( | |
| pipe, message.encode(), len(message), ctypes.byref(bytes_written), None | |
| ) | |
| else: | |
| with open(fifo_file, "w") as file: | |
| file.write(message) | |
| @staticmethod | |
| def read_error_message(fifo_file, pipe=None): | |
| if sys.platform == "win32": | |
| import ctypes | |
| ctypes.windll.kernel32.ConnectNamedPipe(pipe, None) | |
| return read_pipe_message(pipe) | |
| else: | |
| with open(fifo_file, "r") as file: | |
| return file.readline() | |
| @skipIfAsan | |
| def test_runInTerminal(self): | |
| """ | |
| Tests the "runInTerminal" reverse request. It makes sure that the IDE can | |
| launch the inferior with the correct environment variables and arguments. | |
| """ | |
| program = self.getBuildArtifact("a.out") | |
| source = "main.c" | |
| self.build_and_launch( | |
| program, console="integratedTerminal", args=["foobar"], env=["FOO=bar"] | |
| ) | |
| self.dap_server.wait_for_initialized() | |
| self.assertEqual( | |
| len(self.dap_server.reverse_requests), | |
| 1, | |
| "make sure we got a reverse request", | |
| ) | |
| request = self.dap_server.reverse_requests[0] | |
| self.assertIn(self.lldbDAPExec, request["arguments"]["args"]) | |
| self.assertIn(program, request["arguments"]["args"]) | |
| self.assertIn("foobar", request["arguments"]["args"]) | |
| self.assertIn("FOO", request["arguments"]["env"]) | |
| breakpoint_line = line_number(source, "// breakpoint") | |
| self.set_source_breakpoints(source, [breakpoint_line]) | |
| self.continue_to_next_stop() | |
| # We verify we actually stopped inside the loop | |
| counter = int(self.dap_server.get_local_variable_value("counter")) | |
| self.assertEqual(counter, 1) | |
| # We verify we were able to set the launch arguments | |
| argc = int(self.dap_server.get_local_variable_value("argc")) | |
| self.assertEqual(argc, 2) | |
| argv1 = self.dap_server.request_evaluate("argv[1]")["body"]["result"] | |
| self.assertIn("foobar", argv1) | |
| # We verify we were able to set the environment | |
| env = self.dap_server.request_evaluate("foo")["body"]["result"] | |
| self.assertIn("bar", env) | |
| self.continue_to_exit() | |
| @skipIfAsan | |
| def test_runInTerminalWithObjectEnv(self): | |
| """ | |
| Tests the "runInTerminal" reverse request. It makes sure that the IDE can | |
| launch the inferior with the correct environment variables using an object. | |
| """ | |
| program = self.getBuildArtifact("a.out") | |
| self.build_and_launch(program, console="integratedTerminal", env={"FOO": "BAR"}) | |
| self.dap_server.wait_for_initialized() | |
| self.assertEqual( | |
| len(self.dap_server.reverse_requests), | |
| 1, | |
| "make sure we got a reverse request", | |
| ) | |
| request = self.dap_server.reverse_requests[0] | |
| request_envs = request["arguments"]["env"] | |
| self.assertIn("FOO", request_envs) | |
| self.assertEqual("BAR", request_envs["FOO"]) | |
| self.continue_to_exit() | |
| def test_runInTerminalInvalidTarget(self): | |
| self.build_and_create_debug_adapter() | |
| response = self.launch_and_configurationDone( | |
| "INVALIDPROGRAM", | |
| console="integratedTerminal", | |
| args=["foobar"], | |
| env=["FOO=bar"], | |
| ) | |
| self.assertFalse(response["success"]) | |
| self.assertIn( | |
| "'INVALIDPROGRAM' does not exist", | |
| response["body"]["error"]["format"], | |
| ) | |
| def test_missingArgInRunInTerminalLauncher(self): | |
| proc = subprocess.run( | |
| [self.lldbDAPExec, "--launch-target", "INVALIDPROGRAM"], | |
| capture_output=True, | |
| universal_newlines=True, | |
| ) | |
| self.assertNotEqual(proc.returncode, 0) | |
| self.assertIn( | |
| '"--launch-target" requires "--comm-file" to be specified', proc.stderr | |
| ) | |
| def test_FakeAttachedRunInTerminalLauncherWithInvalidProgram(self): | |
| with fifo(directory=self.getBuildDir()) as (comm_file, pipe): | |
| proc = subprocess.Popen( | |
| [ | |
| self.lldbDAPExec, | |
| "--comm-file", | |
| comm_file, | |
| "--launch-target", | |
| "INVALIDPROGRAM", | |
| ], | |
| universal_newlines=True, | |
| stderr=subprocess.PIPE, | |
| ) | |
| if sys.platform == "win32": | |
| _, stderr = proc.communicate() | |
| self.assertIn("Failed to launch target process", stderr) | |
| else: | |
| self.read_pid_message(comm_file, pipe) | |
| self.send_did_attach_message(comm_file, pipe) | |
| self.assertIn( | |
| "No such file or directory", | |
| self.read_error_message(comm_file, pipe), | |
| ) | |
| _, stderr = proc.communicate() | |
| self.assertIn("No such file or directory", stderr) | |
| def test_FakeAttachedRunInTerminalLauncherWithValidProgram(self): | |
| with fifo(directory=self.getBuildDir()) as (comm_file, pipe): | |
| proc = subprocess.Popen( | |
| [ | |
| self.lldbDAPExec, | |
| "--comm-file", | |
| comm_file, | |
| "--launch-target", | |
| "echo", | |
| "foo", | |
| ], | |
| universal_newlines=True, | |
| stdout=subprocess.PIPE, | |
| ) | |
| self.read_pid_message(comm_file, pipe) | |
| self.send_did_attach_message(comm_file, pipe) | |
| stdout, _ = proc.communicate() | |
| self.assertIn("foo", stdout) | |
| def test_FakeAttachedRunInTerminalLauncherAndCheckEnvironment(self): | |
| with fifo(directory=self.getBuildDir()) as (comm_file, pipe): | |
| proc = subprocess.Popen( | |
| [self.lldbDAPExec, "--comm-file", comm_file, "--launch-target", "env"], | |
| universal_newlines=True, | |
| stdout=subprocess.PIPE, | |
| env={**os.environ, "FOO": "BAR"}, | |
| ) | |
| self.read_pid_message(comm_file, pipe) | |
| self.send_did_attach_message(comm_file, pipe) | |
| stdout, _ = proc.communicate() | |
| self.assertIn("FOO=BAR", stdout) | |
| def test_NonAttachedRunInTerminalLauncher(self): | |
| with fifo(directory=self.getBuildDir()) as (comm_file, pipe): | |
| proc = subprocess.Popen( | |
| [ | |
| self.lldbDAPExec, | |
| "--comm-file", | |
| comm_file, | |
| "--launch-target", | |
| "echo", | |
| "foo", | |
| ], | |
| universal_newlines=True, | |
| stderr=subprocess.PIPE, | |
| env={**os.environ, "LLDB_DAP_RIT_TIMEOUT_IN_MS": "1000"}, | |
| ) | |
| self.read_pid_message(comm_file, pipe) | |
| _, stderr = proc.communicate() | |
| self.assertIn("Timed out trying to get messages from the debug adapter", stderr) | |
| def test_client_missing_runInTerminal_feature(self): | |
| program = self.getBuildArtifact("a.out") | |
| self.build_and_create_debug_adapter() | |
| response = self.launch_and_configurationDone( | |
| program, | |
| console="integratedTerminal", | |
| client_features={"supportsRunInTerminalRequest": False}, | |
| ) | |
| self.assertFalse(response["success"], f"Expected failure got {response!r}") | |
| error_message = response["body"]["error"]["format"] | |
| self.assertIn("Client does not support RunInTerminal.", error_message) |