[Lldb-commits] [lldb] [lldb-dap] Migrate threads and disassemble DAP test (PR #211277)

via lldb-commits lldb-commits at lists.llvm.org
Wed Jul 22 13:48:00 PDT 2026


llvmorg-github-actions[bot] wrote:


<!--LLVM PR SUMMARY COMMENT-->

@llvm/pr-subscribers-lldb

Author: Ebuka Ezike (da-viper)

<details>
<summary>Changes</summary>



---
Full diff: https://github.com/llvm/llvm-project/pull/211277.diff


2 Files Affected:

- (modified) lldb/test/API/tools/lldb-dap/disassemble/TestDAP_disassemble.py (+65-58) 
- (modified) lldb/test/API/tools/lldb-dap/threads/TestDAP_threads.py (+52-46) 


``````````diff
diff --git a/lldb/test/API/tools/lldb-dap/disassemble/TestDAP_disassemble.py b/lldb/test/API/tools/lldb-dap/disassemble/TestDAP_disassemble.py
index 6c41c86ff9ae5..db6847da987fb 100644
--- a/lldb/test/API/tools/lldb-dap/disassemble/TestDAP_disassemble.py
+++ b/lldb/test/API/tools/lldb-dap/disassemble/TestDAP_disassemble.py
@@ -4,47 +4,50 @@
 
 from lldbsuite.test.decorators import skipIfWindows
 from lldbsuite.test.lldbtest import line_number
-import lldbdap_testcase
+from lldbsuite.test.tools.lldb_dap.types import LaunchArgs
+from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase
 
 
-class TestDAP_disassemble(lldbdap_testcase.DAPTestCaseBase):
+class TestDAP_disassemble(DAPTestCaseBase):
     @skipIfWindows
     def test_disassemble(self):
-        """
-        Tests the 'disassemble' request.
-        """
+        """Disassembly at the current PC returns the expected source line, and
+        clearing breakpoints doesn't change the instructions."""
+
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-        source = "main.c"
-        bp_line_no = line_number(source, "// breakpoint 1")
-        self.set_source_breakpoints(source, [bp_line_no])
-        self.continue_to_next_stop()
-
-        insts_with_bp, pc_with_bp_assembly = self.disassemble(frameIndex=0)
-        self.assertIn("location", pc_with_bp_assembly, "Source location missing.")
-        self.assertEqual(
-            pc_with_bp_assembly["line"], bp_line_no, "Expects the same line number"
-        )
-        no_bp = self.set_source_breakpoints(source, [])
-        self.assertEqual(len(no_bp), 0, "Expects no breakpoints.")
-        self.assertIn(
-            "instruction", pc_with_bp_assembly, "Assembly instruction missing."
-        )
+        session = self.build_and_create_session()
+        source = self.getSourcePath("main.c")
+        bp_line = line_number(source, "// breakpoint 1")
+
+        with session.configure(LaunchArgs(program)) as ctx:
+            session.resolve_source_breakpoints(source, [bp_line])
+        stop_event = session.verify_stopped_on_breakpoint(after=ctx.process_event)
+        top_frame = session.thread_context_from(stop_event).top_frame()
+
+        insts_with_bp = top_frame.disassemble()
+        pc_with_bp = insts_with_bp[0]
+        self.assertIsNotNone(pc_with_bp.location, "Source location missing.")
+        self.assertEqual(pc_with_bp.line, bp_line, "Expects the same line number")
+        self.assertTrue(pc_with_bp.instruction, "Assembly instruction missing.")
 
-        insts_no_bp, pc_no_bp_assembly = self.disassemble(frameIndex=0)
-        self.assertIn("location", pc_no_bp_assembly, "Source location missing.")
+        cleared = session.set_source_breakpoints(source, [])
+        self.assertEqual(len(cleared.body.breakpoints), 0, "Expects no breakpoints.")
+
+        insts_no_bp = top_frame.disassemble()
+        pc_no_bp = insts_no_bp[0]
+        self.assertIsNotNone(pc_no_bp.location, "Source location missing.")
+        self.assertEqual(pc_no_bp.line, bp_line, "Expects the same line number")
+        self.assertTrue(pc_no_bp.instruction, "Assembly instruction missing.")
+
+        # The disassembly instructions should be the same with breakpoint and
+        # no breakpoint.
         self.assertEqual(
-            pc_with_bp_assembly["line"], bp_line_no, "Expects the same line number"
-        )
-        # the disassembly instructions should be the same with breakpoint and no breakpoint;
-        self.assertDictEqual(
             insts_with_bp,
             insts_no_bp,
             "Expects instructions are the same after removing breakpoints.",
         )
-        self.assertIn("instruction", pc_no_bp_assembly, "Assembly instruction missing.")
 
-        self.continue_to_exit()
+        session.continue_to_exit()
 
     @skipIfWindows
     def test_disassemble_backwards(self):
@@ -52,18 +55,22 @@ def test_disassemble_backwards(self):
         Tests the 'disassemble' request with a backwards disassembly range.
         """
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-        source = "main.c"
-        self.set_source_breakpoints(source, [line_number(source, "// breakpoint 1")])
-        self.continue_to_next_stop()
-
-        instruction_pointer_reference = self.get_stackFrames()[1][
-            "instructionPointerReference"
-        ]
+        session = self.build_and_create_session()
+        source = self.getSourcePath("main.c")
+        bp_line = line_number(source, "// breakpoint 1")
+        with session.configure(LaunchArgs(program)) as ctx:
+            session.resolve_source_breakpoints(source, [bp_line])
+        stop_event = session.verify_stopped_on_breakpoint(after=ctx.process_event)
+
+        caller_frame = session.thread_context_from(stop_event).frames(levels=2)[1]
+        instruction_pointer_ref = self.expect_not_none(
+            caller_frame.frame.instructionPointerReference
+        )
+
         backwards_instructions = 200
         instructions_count = 400
-        instructions = self.dap_server.request_disassemble(
-            memoryReference=instruction_pointer_reference,
+        instructions = session.disassemble(
+            memoryReference=instruction_pointer_ref,
             instructionOffset=-backwards_instructions,
             instructionCount=instructions_count,
         )
@@ -77,39 +84,39 @@ def test_disassemble_backwards(self):
         frame_instruction_index = next(
             (
                 i
-                for i, instruction in enumerate(instructions)
-                if instruction["address"] == instruction_pointer_reference
+                for i, inst in enumerate(instructions)
+                if inst.address == instruction_pointer_ref
             ),
             -1,
         )
         self.assertEqual(
             frame_instruction_index,
             backwards_instructions,
-            f"requested instruction should be preceeded by {backwards_instructions} instructions. Actual index: {frame_instruction_index}",
+            f"requested instruction should be preceded by {backwards_instructions} "
+            f"instructions. Actual index: {frame_instruction_index}",
         )
 
-        # clear breakpoints
-        self.set_source_breakpoints(source, [])
-        self.continue_to_exit()
+        session.set_source_breakpoints(source, [])
+        session.continue_to_exit()
 
     def test_disassemble_empty_memory_reference(self):
-        """
-        Tests the 'disassemble' request with empty memory reference.
-        """
+        """An empty `memoryReference` returns the requested count of invalid
+        placeholder instructions instead of erroring out."""
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-        source = "main.c"
-        bp_line_no = line_number(source, "// breakpoint 1")
-        self.set_source_breakpoints(source, [bp_line_no])
-        self.continue_to_next_stop()
+        session = self.build_and_create_session()
+        source = self.getSourcePath("main.c")
+        bp_line = line_number(source, "// breakpoint 1")
+        with session.configure(LaunchArgs(program)) as ctx:
+            session.resolve_source_breakpoints(source, [bp_line])
+        session.verify_stopped_on_breakpoint(after=ctx.process_event)
 
-        instructions = self.dap_server.request_disassemble(
+        instructions = session.disassemble(
             memoryReference="", instructionOffset=0, instructionCount=50
         )
         self.assertEqual(len(instructions), 50)
         for instruction in instructions:
-            self.assertEqual(instruction["presentationHint"], "invalid")
+            self.assertEqual(instruction.presentationHint, "invalid")
 
-        # clear breakpoints
-        self.set_source_breakpoints(source, [])
-        self.continue_to_exit()
+        # Clear breakpoints and exit.
+        session.set_source_breakpoints(source, [])
+        session.continue_to_exit()
diff --git a/lldb/test/API/tools/lldb-dap/threads/TestDAP_threads.py b/lldb/test/API/tools/lldb-dap/threads/TestDAP_threads.py
index 79a35e1433275..05e87e3f437c6 100644
--- a/lldb/test/API/tools/lldb-dap/threads/TestDAP_threads.py
+++ b/lldb/test/API/tools/lldb-dap/threads/TestDAP_threads.py
@@ -2,14 +2,14 @@
 Test lldb-dap threads request
 """
 
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
+from lldbsuite.test.decorators import skipIfTargetDoesNotSupportThreads
+from lldbsuite.test.lldbtest import line_number
+from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase
+from lldbsuite.test.tools.lldb_dap.types import LaunchArgs, StoppedReason, ThreadsArgs
 
 
 @skipIfTargetDoesNotSupportThreads()
-class TestDAP_threads(lldbdap_testcase.DAPTestCaseBase):
+class TestDAP_threads(DAPTestCaseBase):
     def test_correct_thread(self):
         """
         Tests that the correct thread is selected if we continue from
@@ -17,65 +17,71 @@ def test_correct_thread(self):
         In this case, the selected thread should be the thread that
         just hit the breakpoint, and not the first thread in the list.
         """
+        session = self.build_and_create_session()
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
         source = "main.cpp"
         breakpoint_line = line_number(source, "// break here")
-        lines = [breakpoint_line]
-        # Set breakpoint in the thread function
-        breakpoint_ids = self.set_source_breakpoints(source, lines)
-        self.assertEqual(
-            len(breakpoint_ids), len(lines), "expect correct number of breakpoints"
-        )
-        self.continue_to_breakpoints(breakpoint_ids)
+
+        with session.configure(LaunchArgs(program)) as ctx:
+            breakpoint_ids = session.resolve_source_breakpoints(
+                source, [breakpoint_line]
+            )
+        first_stop = session.verify_stopped_on_breakpoint(after=ctx.process_event)
+
         # We're now stopped at the breakpoint in the first thread, thread #2.
         # Continue to join the first thread and hit the breakpoint in the
         # second thread, thread #3.
-        self.dap_server.request_continue()
-        stopped_event = self.dap_server.wait_for_stopped()
-        # Verify that the description is the relevant breakpoint,
-        # preserveFocusHint is False and threadCausedFocus is True
-        self.assertTrue(
-            stopped_event[0]["body"]["description"].startswith(
-                "breakpoint %s." % breakpoint_ids[0]
-            )
+        second_stop = session.continue_to_next_stop(exp_reason=StoppedReason.BREAKPOINT)
+        self.assertNotEqual(
+            first_stop.body.threadId,
+            second_stop.body.threadId,
+            "the stopped events should be on different threads.",
         )
-        self.assertNotIn("preserveFocusHint", stopped_event[0]["body"])
-        # All threads should be named Thread {index}
-        threads = self.dap_server.get_threads()
-        self.assertTrue(all(len(t["name"]) > 0 for t in threads))
+
+        # Verify that the description is the relevant breakpoint,
+        # preserveFocusHint is False and threadCausedFocus is True.
+        stop_description = self.expect_not_none(second_stop.body.description)
+        self.assertTrue(stop_description.startswith(f"breakpoint {breakpoint_ids[0]}"))
+        self.assertIsNone(second_stop.body.preserveFocusHint)
+
+        # All threads should have a name.
+        threads = session.send_request(ThreadsArgs()).result().body.threads
+        for t in threads:
+            self.assertTrue(t.name, "thread name should be non-empty")
+
+        session.continue_to_exit()
 
     def test_thread_format(self):
-        """
-        Tests the support for custom thread formats.
-        """
+        """Tests the support for custom thread formats."""
+        session = self.build_and_create_session()
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(
-            program,
-            customThreadFormat="This is thread index #${thread.index}",
-            stopCommands=["thread list"],
-        )
         source = "main.cpp"
         breakpoint_line = line_number(source, "// break here")
-        lines = [breakpoint_line]
-        # Set breakpoint in the thread function
-        breakpoint_ids = self.set_source_breakpoints(source, lines)
-        self.assertEqual(
-            len(breakpoint_ids), len(lines), "expect correct number of breakpoints"
-        )
-        self.continue_to_breakpoints(breakpoint_ids)
-        # We are stopped at the first thread
-        threads = self.dap_server.get_threads()
-        print("got thread", threads)
+
+        with session.configure(
+            LaunchArgs(
+                program,
+                customThreadFormat="This is thread index #${thread.index}",
+                stopCommands=["thread list"],
+            )
+        ) as ctx:
+            bp_ids = session.resolve_source_breakpoints(source, [breakpoint_line])
+        session.verify_stopped_on_breakpoint(bp_ids, after=ctx.process_event)
+
+        threads = session.send_request(ThreadsArgs()).result().body.threads
         if self.getPlatform() == "windows":
             # Windows creates a thread pool once WaitForSingleObject is called
             # by thread.join(). As we are in the thread function, we can't be
             # certain that join() has been called yet and a thread pool has
             # been created, thus we only check for the first two threads.
-            names = list(sorted(t["name"] for t in threads))[:2]
+            names = sorted(t.name for t in threads)[:2]
             self.assertEqual(
                 names, ["This is thread index #1", "This is thread index #2"]
             )
         else:
-            self.assertEqual(threads[0]["name"], "This is thread index #1")
-            self.assertEqual(threads[1]["name"], "This is thread index #2")
+            self.assertEqual(threads[0].name, "This is thread index #1")
+            self.assertEqual(threads[1].name, "This is thread index #2")
+
+        # Clear the breakpoint so the second thread doesn't hit it on the way out.
+        session.set_source_breakpoints(source, [])
+        session.continue_to_exit()

``````````

</details>


https://github.com/llvm/llvm-project/pull/211277


More information about the lldb-commits mailing list