Author: Ebuka Ezike Date: 2026-07-22T15:40:16+01:00 New Revision: fed8a6f2e38bd82835ddb651305782db6b62f326
URL: https://github.com/llvm/llvm-project/commit/fed8a6f2e38bd82835ddb651305782db6b62f326 DIFF: https://github.com/llvm/llvm-project/commit/fed8a6f2e38bd82835ddb651305782db6b62f326.diff LOG: [lldb-dap] Migrate DAP cancel and commands tests (#211044) Added: Modified: lldb/test/API/tools/lldb-dap/cancel/TestDAP_cancel.py lldb/test/API/tools/lldb-dap/commands/TestDAP_commands.py Removed: ################################################################################ diff --git a/lldb/test/API/tools/lldb-dap/cancel/TestDAP_cancel.py b/lldb/test/API/tools/lldb-dap/cancel/TestDAP_cancel.py index 67b63df1951f1..f639d01f851b2 100644 --- a/lldb/test/API/tools/lldb-dap/cancel/TestDAP_cancel.py +++ b/lldb/test/API/tools/lldb-dap/cancel/TestDAP_cancel.py @@ -2,99 +2,92 @@ Test lldb-dap cancel request """ -import os import time -import lldbdap_testcase +from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase +from lldbsuite.test.tools.lldb_dap.session_helpers import DAPTestSession +from lldbsuite.test.tools.lldb_dap.types import CancelArgs, EvaluateArgs, LaunchArgs -class TestDAP_cancel(lldbdap_testcase.DAPTestCaseBase): - def send_async_req(self, command: str, arguments: dict = {}) -> int: - return self.dap_server.send_packet( - { - "seq": 0, - "type": "request", - "command": command, - "arguments": arguments, - } - ) - def async_blocking_request(self, count: int) -> int: +class TestDAP_cancel(DAPTestCaseBase): + def async_blocking_request(self, session: DAPTestSession, count: int): """ Sends an evaluate request that will sleep for the specified count to block the request handling thread. """ - return self.send_async_req( - command="evaluate", - arguments={"expression": f"`busy-loop {count}", "context": "repl"}, + return session.send_request( + EvaluateArgs(expression=f"`busy-loop {count}", context="repl") ) - def async_cancel(self, requestId: int) -> int: - return self.send_async_req(command="cancel", arguments={"requestId": requestId}) + def async_cancel(self, session: DAPTestSession, requestId: int): + return session.send_request(CancelArgs(requestId=requestId)) def test_pending_request(self): - """ - Tests cancelling a pending request. - """ + """Tests cancelling a pending request.""" program = self.getBuildArtifact("a.out") busy_loop = self.getSourcePath("busy_loop.py") - self.build_and_launch( - program, - initCommands=[f"command script import {busy_loop}"], - stopOnEntry=True, + session = self.build_and_create_session() + process_event = session.launch( + LaunchArgs( + program, + initCommands=[f"command script import {busy_loop}"], + stopOnEntry=True, + ) ) - self.verify_stop_on_entry() + session.verify_stopped_on_entry(after=process_event) # Use a relatively short timeout since this is only to ensure the # following request is queued. - blocking_seq = self.async_blocking_request(count=1) + blocking_handle = self.async_blocking_request(session, count=1) # Use a longer timeout to ensure we catch if the request was interrupted # properly. - pending_seq = self.async_blocking_request(count=10) - cancel_seq = self.async_cancel(requestId=pending_seq) + pending_handle = self.async_blocking_request(session, count=10) + cancel_handle = self.async_cancel(session, requestId=pending_handle.seq) - blocking_resp = self.dap_server.receive_response(blocking_seq) - self.assertEqual(blocking_resp["request_seq"], blocking_seq) - self.assertEqual(blocking_resp["command"], "evaluate") - self.assertEqual(blocking_resp["success"], True) + blocking_resp = blocking_handle.result() + self.assertEqual(blocking_resp.request_seq, blocking_handle.seq) + self.assertEqual(blocking_resp.command, "evaluate") + self.assertEqual(blocking_resp.success, True) - pending_resp = self.dap_server.receive_response(pending_seq) - self.assertEqual(pending_resp["request_seq"], pending_seq) - self.assertEqual(pending_resp["command"], "evaluate") - self.assertEqual(pending_resp["success"], False) - self.assertEqual(pending_resp["message"], "cancelled") + pending_resp = pending_handle.error() + self.assertEqual(pending_resp.request_seq, pending_handle.seq) + self.assertEqual(pending_resp.command, "evaluate") + self.assertEqual(pending_resp.success, False) + self.assertEqual(pending_resp.message, "cancelled") - cancel_resp = self.dap_server.receive_response(cancel_seq) - self.assertEqual(cancel_resp["request_seq"], cancel_seq) - self.assertEqual(cancel_resp["command"], "cancel") - self.assertEqual(cancel_resp["success"], True) - self.continue_to_exit() + cancel_resp = cancel_handle.result() + self.assertEqual(cancel_resp.request_seq, cancel_handle.seq) + self.assertEqual(cancel_resp.command, "cancel") + self.assertEqual(cancel_resp.success, True) + session.continue_to_exit() def test_inflight_request(self): - """ - Tests cancelling an inflight request. - """ + """Tests cancelling an inflight request.""" program = self.getBuildArtifact("a.out") busy_loop = self.getSourcePath("busy_loop.py") - self.build_and_launch( - program, - initCommands=[f"command script import {busy_loop}"], - stopOnEntry=True, + session = self.build_and_create_session() + process_event = session.launch( + LaunchArgs( + program, + initCommands=[f"command script import {busy_loop}"], + stopOnEntry=True, + ) ) - self.verify_stop_on_entry() + session.verify_stopped_on_entry(after=process_event) - blocking_seq = self.async_blocking_request(count=10) + blocking_handle = self.async_blocking_request(session, count=10) # Wait for the sleep to start to cancel the inflight request. time.sleep(0.5) - cancel_seq = self.async_cancel(requestId=blocking_seq) + cancel_handle = self.async_cancel(session, requestId=blocking_handle.seq) - blocking_resp = self.dap_server.receive_response(blocking_seq) - self.assertEqual(blocking_resp["request_seq"], blocking_seq) - self.assertEqual(blocking_resp["command"], "evaluate") - self.assertEqual(blocking_resp["success"], False) - self.assertEqual(blocking_resp["message"], "cancelled") + blocking_resp = blocking_handle.error() + self.assertEqual(blocking_resp.request_seq, blocking_handle.seq) + self.assertEqual(blocking_resp.command, "evaluate") + self.assertEqual(blocking_resp.success, False) + self.assertEqual(blocking_resp.message, "cancelled") - cancel_resp = self.dap_server.receive_response(cancel_seq) - self.assertEqual(cancel_resp["request_seq"], cancel_seq) - self.assertEqual(cancel_resp["command"], "cancel") - self.assertEqual(cancel_resp["success"], True) - self.continue_to_exit() + cancel_resp = cancel_handle.result() + self.assertEqual(cancel_resp.request_seq, cancel_handle.seq) + self.assertEqual(cancel_resp.command, "cancel") + self.assertEqual(cancel_resp.success, True) + session.continue_to_exit() diff --git a/lldb/test/API/tools/lldb-dap/commands/TestDAP_commands.py b/lldb/test/API/tools/lldb-dap/commands/TestDAP_commands.py index e844a0e929ed9..e79c2895fa963 100644 --- a/lldb/test/API/tools/lldb-dap/commands/TestDAP_commands.py +++ b/lldb/test/API/tools/lldb-dap/commands/TestDAP_commands.py @@ -2,60 +2,64 @@ Test lldb-dap command hooks """ -import lldbdap_testcase -from lldbsuite.test.decorators import * +from lldbsuite.test.tools.lldb_dap.types import AttachArgs, LaunchArgs +from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase -class TestDAP_commands(lldbdap_testcase.DAPTestCaseBase): - SHARED_BUILD_TESTCASE = False - +class TestDAP_commands(DAPTestCaseBase): def test_command_directive_quiet_on_success(self): program = self.getBuildArtifact("a.out") - command_quiet = ( + quiet_command = ( "settings set target.show-hex-variable-values-with-leading-zeroes false" ) - command_not_quiet = ( + visible_command = ( "settings set target.show-hex-variable-values-with-leading-zeroes true" ) - self.build_and_launch( - program, - initCommands=["?" + command_quiet, command_not_quiet], - terminateCommands=["?" + command_quiet, command_not_quiet], - stopCommands=["?" + command_quiet, command_not_quiet], - exitCommands=["?" + command_quiet, command_not_quiet], - ) - full_output = self.collect_console( - pattern=command_not_quiet, + commands = [f"?{quiet_command}", visible_command] + session = self.build_and_create_session() + process_event = session.launch( + LaunchArgs( + program, + initCommands=commands, + terminateCommands=commands, + stopCommands=commands, + exitCommands=commands, + ) ) - self.assertNotIn(command_quiet, full_output) - self.assertIn(command_not_quiet, full_output) + session.verify_process_exited(after=process_event) + full_output = session.get_console() + self.assertNotIn(quiet_command, full_output) + self.assertIn(visible_command, full_output) def do_test_abort_on_error( self, - use_init_commands=False, - use_launch_commands=False, - use_pre_run_commands=False, - use_post_run_commands=False, + use_init_commands: bool = False, + use_launch_commands: bool = False, + use_pre_run_commands: bool = False, + use_post_run_commands: bool = False, ): program = self.getBuildArtifact("a.out") - command_quiet = ( + quiet_command = ( "settings set target.show-hex-variable-values-with-leading-zeroes false" ) - command_abort_on_error = "settings set foo bar" - commands = ["?!" + command_quiet, "!" + command_abort_on_error] - self.build_and_launch( - program, - initCommands=commands if use_init_commands else None, - launchCommands=commands if use_launch_commands else None, - preRunCommands=commands if use_pre_run_commands else None, - postRunCommands=commands if use_post_run_commands else None, - ) - self.verify_configuration_done(use_post_run_commands) - full_output = self.collect_console( - pattern=command_abort_on_error, + fake_command = "settings set foo bar" + commands = [f"?!{quiet_command}", f"!{fake_command}"] + + session = self.build_and_create_session() + pending_response = session.initialize_and_launch( + LaunchArgs( + program, + initCommands=commands if use_init_commands else None, + launchCommands=commands if use_launch_commands else None, + preRunCommands=commands if use_pre_run_commands else None, + postRunCommands=commands if use_post_run_commands else None, + ) ) - self.assertNotIn(command_quiet, full_output) - self.assertIn(command_abort_on_error, full_output) + session.verify_configuration_done(use_post_run_commands) + pending_response.result_or_error() + full_output = session.get_console() + self.assertNotIn(quiet_command, full_output) + self.assertIn(fake_command, full_output) def test_command_directive_abort_on_error_init_commands(self): self.do_test_abort_on_error(use_init_commands=True) @@ -70,16 +74,20 @@ def test_command_directive_abort_on_error_post_run_commands(self): self.do_test_abort_on_error(use_post_run_commands=True) def test_command_directive_abort_on_error_attach_commands(self): - command_quiet = ( + program = self.getBuildArtifact("a.out") + quiet_command = ( "settings set target.show-hex-variable-values-with-leading-zeroes false" ) - command_abort_on_error = "settings set foo bar" - program = self.build_and_create_debug_adapter_for_attach() - resp = self.attach_and_configurationDone( + fake_command = "settings set foo bar" + session = self.build_and_create_session() + session.initialize_sequence(session.initialize_args) + attach_args = AttachArgs( program=program, - attachCommands=["?!" + command_quiet, "!" + command_abort_on_error], + attachCommands=[f"?!{quiet_command}", f"!{fake_command}"], ) - self.assertFalse(resp["success"], "expected 'attach' failure") - full_output = self.collect_console(pattern=command_abort_on_error) - self.assertNotIn(command_quiet, full_output) - self.assertIn(command_abort_on_error, full_output) + with self.assertRaises(AssertionError): + session.attach(attach_args) + + full_output = session.get_console() + self.assertNotIn(quiet_command, full_output) + self.assertIn(fake_command, full_output) _______________________________________________ lldb-commits mailing list [email protected] https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
