diff --git a/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/lldbdap_testcase.py b/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/lldbdap_testcase.py index c51b4b1892951..36471501ea666 100644 --- a/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/lldbdap_testcase.py +++ b/lldb/packages/Python/lldbsuite/test/tools/lldb-dap/lldbdap_testcase.py @@ -1,15 +1,16 @@ import os import time -from typing import Optional import uuid +from typing import Optional import dap_server from dap_server import Source from lldbsuite.test.lldbtest import * -from lldbsuite.test import lldbplatformutil -import lldbgdbserverutils import base64 +import lldbgdbserverutils +from lldbsuite.test import lldbplatformutil + class DAPTestCaseBase(TestBase): # set timeout based on whether ASAN was enabled or not. Increase @@ -451,8 +452,25 @@ def attach( # if we throw an exception during the test case. def cleanup(): if disconnectAutomatically: - self.dap_server.request_disconnect(terminateDebuggee=True) - self.dap_server.terminate() + try: + self.dap_server.request_disconnect(terminateDebuggee=True) + except ( + ValueError, + TimeoutError, + BrokenPipeError, + ConnectionError, + Exception, + ) as e: + # DAP server might not be responsive, skip disconnect and terminate directly + print( + f"Warning: disconnect failed ({e}), skipping and terminating directly" + ) + try: + self.dap_server.terminate() + except Exception as e: + print( + f"Warning: terminate failed ({e}), DAP server may have already died" + ) # Execute the cleanup function during test case tear down. self.addTearDownHook(cleanup) @@ -462,9 +480,20 @@ def cleanup(): if expectFailure: return response if not (response and response["success"]): - self.assertTrue( - response["success"], "attach failed (%s)" % (response["message"]) - ) + error_msg = "attach failed" + if response: + if "message" in response: + error_msg += " (%s)" % response["message"] + elif "body" in response and "error" in response["body"]: + if "format" in response["body"]["error"]: + error_msg += " (%s)" % response["body"]["error"]["format"] + else: + error_msg += " (error in body)" + else: + error_msg += " (no error details available)" + else: + error_msg += " (no response)" + self.assertTrue(response and response["success"], error_msg) def launch( self, diff --git a/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py b/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py index 55557e6e0030e..67d3ffaf16250 100644 --- a/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py +++ b/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py @@ -4,12 +4,13 @@ from lldbsuite.test.decorators import * from lldbsuite.test.lldbtest import * -from lldbsuite.test import lldbutil -import lldbdap_testcase import subprocess import threading import time +import lldbdap_testcase +from lldbsuite.test import lldbutil + def spawn_and_wait(program, delay): if delay: @@ -227,3 +228,46 @@ def test_terminate_commands(self): pattern=terminateCommands[0], ) self.verify_commands("terminateCommands", output, terminateCommands) + + def test_session_id_update(self): + program = self.build_and_create_debug_adapter_for_attach() + self.process = subprocess.Popen( + [program], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + postRunCommands = [ + "script print('Actual_Session_ID: ' + str(os.getenv('VSCODE_DEBUG_SESSION_ID', 'None')))" + ] + self.attach( + pid=self.process.pid, + postRunCommands=postRunCommands, + ) + output = self.get_console() + lines = filter(lambda x: "Actual_Session_ID" in x, output.splitlines()) + self.assertTrue( + any("Actual_Session_ID: None" in l for l in lines), + "expect session id to be None when not set", + ) + + def test_session_id_update_empty(self): + program = self.build_and_create_debug_adapter_for_attach() + + self.spawn_thread = threading.Thread( + target=spawn_and_wait, + args=(program, 0.1), + ) + self.spawn_thread.start() + + postRunCommands = [ + "script print('Actual_Session_ID: ' + str(os.getenv('VSCODE_DEBUG_SESSION_ID', 'None')))" + ] + self.attach(program=program, postRunCommands=postRunCommands) + output = self.get_console() + lines = filter(lambda x: "Actual_Session_ID" in x, output.splitlines()) + self.assertTrue( + any("Actual_Session_ID: None" in l for l in lines), + "expect session id in console output", + )