[Lldb-commits] [lldb] fed8a6f - [lldb-dap] Migrate DAP cancel and commands tests (#211044)
via lldb-commits
lldb-commits at lists.llvm.org
Wed Jul 22 07:40:22 PDT 2026
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)
More information about the lldb-commits
mailing list