https://github.com/da-viper created 
https://github.com/llvm/llvm-project/pull/208166

migrate breakpointLocations, logpoints, setBreakpoints exceptionBreakpoints and 
functionBreakpoints test

>From e7d4d3948291bd0310dd0423af8063e7ea45c0a6 Mon Sep 17 00:00:00 2001
From: Ebuka Ezike <[email protected]>
Date: Tue, 7 Jul 2026 17:30:40 +0100
Subject: [PATCH] [lldb-dap] Migrate lldb-dap tests in `lldb-dap/breakpoint`

migrate breakpointLocations, logpoints, setBreakpoints
exceptionBreakpoints and functionBreakpoints test
---
 .../breakpoint/TestDAP_breakpointLocations.py | 121 ++---
 .../lldb-dap/breakpoint/TestDAP_logpoints.py  | 310 +++++-------
 .../breakpoint/TestDAP_setBreakpoints.py      | 477 +++++++++---------
 .../TestDAP_setExceptionBreakpoints.py        |  35 +-
 .../TestDAP_setFunctionBreakpoints.py         | 189 ++++---
 5 files changed, 531 insertions(+), 601 deletions(-)

diff --git 
a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_breakpointLocations.py 
b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_breakpointLocations.py
index 1fe5f8b9e2adc..f97b4b403b063 100644
--- a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_breakpointLocations.py
+++ b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_breakpointLocations.py
@@ -2,88 +2,77 @@
 Test lldb-dap breakpointLocations request
 """
 
-
-import dap_server
-import shutil
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
 import os
 
+from lldbsuite.test.decorators import (
+    skipIfTargetDoesNotSupportSharedLibraries,
+    skipIfWindows,
+)
+from lldbsuite.test.lldbtest import line_number
+from lldbsuite.test.tools.lldb_dap.dap_types import BreakpointLocation, 
LaunchArgs
+from lldbsuite.test.tools.lldb_dap.lldb_dap_testcase import DAPTestCaseBase
 
-@skipIfTargetDoesNotSupportSharedLibraries()
-class TestDAP_breakpointLocations(lldbdap_testcase.DAPTestCaseBase):
-    def setUp(self):
-        lldbdap_testcase.DAPTestCaseBase.setUp(self)
-
-        self.main_basename = "main-copy.cpp"
-        self.main_path = 
os.path.realpath(self.getBuildArtifact(self.main_basename))
 
+@skipIfTargetDoesNotSupportSharedLibraries()
+class TestDAP_breakpointLocations(DAPTestCaseBase):
     @skipIfWindows
     def test_column_breakpoints(self):
         """Test retrieving the available breakpoint locations."""
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program, stopOnEntry=True)
-        loop_line = line_number(self.main_path, "// break loop")
-        self.dap_server.request_continue()
+        session = self.build_and_create_session()
+        main_path = os.path.realpath(self.getBuildArtifact("main-copy.cpp"))
 
-        # Ask for the breakpoint locations based only on the line number
-        response = self.dap_server.request_breakpointLocations(
-            self.main_path, loop_line
-        )
-        self.assertTrue(response["success"])
-        self.assertEqual(
-            response["body"]["breakpoints"],
-            [
-                {"line": loop_line, "column": 9},
-                {"line": loop_line, "column": 13},
-                {"line": loop_line, "column": 20},
-                {"line": loop_line, "column": 23},
-                {"line": loop_line, "column": 25},
-                {"line": loop_line, "column": 34},
-                {"line": loop_line, "column": 37},
-                {"line": loop_line, "column": 39},
-                {"line": loop_line, "column": 51},
-            ],
-        )
+        process_event = session.launch(LaunchArgs(program, stopOnEntry=True))
+        session.verify_stopped_on_entry(after=process_event)
 
-        # Ask for the breakpoint locations for a column range
-        response = self.dap_server.request_breakpointLocations(
-            self.main_path,
-            loop_line,
-            column=24,
-            end_column=46,
-        )
-        self.assertTrue(response["success"])
-        self.assertEqual(
-            response["body"]["breakpoints"],
-            [
-                {"line": loop_line, "column": 25},
-                {"line": loop_line, "column": 34},
-                {"line": loop_line, "column": 37},
-                {"line": loop_line, "column": 39},
-            ],
+        # Ask for the breakpoint locations based only on the line number.
+        loop_line = line_number(main_path, "// break loop")
+        response = session.get_breakpoint_locations(main_path, loop_line)
+        breakpoint_locations = response.body.breakpoints
+
+        expected_locations = [
+            BreakpointLocation(line=loop_line, column=9),
+            BreakpointLocation(line=loop_line, column=13),
+            BreakpointLocation(line=loop_line, column=20),
+            BreakpointLocation(line=loop_line, column=23),
+            BreakpointLocation(line=loop_line, column=25),
+            BreakpointLocation(line=loop_line, column=34),
+            BreakpointLocation(line=loop_line, column=37),
+            BreakpointLocation(line=loop_line, column=39),
+            BreakpointLocation(line=loop_line, column=51),
+        ]
+        self.assertEqual(breakpoint_locations, expected_locations)
+
+        # Ask for the breakpoint locations for a column range.
+        response = session.get_breakpoint_locations(
+            main_path, loop_line, column=24, endColumn=46
         )
+        breakpoint_locations = response.body.breakpoints
+        expected_locations = [
+            BreakpointLocation(line=loop_line, column=25),
+            BreakpointLocation(line=loop_line, column=34),
+            BreakpointLocation(line=loop_line, column=37),
+            BreakpointLocation(line=loop_line, column=39),
+        ]
+        self.assertEqual(breakpoint_locations, expected_locations)
 
-        # Ask for the breakpoint locations for a range of line numbers
-        response = self.dap_server.request_breakpointLocations(
-            self.main_path,
-            line=loop_line,
-            end_line=loop_line + 2,
-            column=39,
+        # Ask for the breakpoint locations for a range of line numbers.
+        response = session.get_breakpoint_locations(
+            main_path, line=loop_line, column=39, endLine=loop_line + 2
         )
         self.maxDiff = None
-        self.assertTrue(response["success"])
         # On some systems, there is an additional breakpoint available
         # at loop_line + 1, column 3, i.e. at the end of the loop. To make
         # this test more portable, only check that all expected breakpoints
         # are presented, but also accept additional breakpoints.
-        expected_breakpoints = [
-            {"column": 39, "line": loop_line},
-            {"column": 51, "line": loop_line},
-            {"column": 3, "line": loop_line + 2},
-            {"column": 18, "line": loop_line + 2},
+        expected_locations = [
+            BreakpointLocation(line=loop_line, column=39),
+            BreakpointLocation(line=loop_line, column=51),
+            BreakpointLocation(line=loop_line + 2, column=3),
+            BreakpointLocation(line=loop_line + 2, column=18),
         ]
-        for bp in expected_breakpoints:
-            self.assertIn(bp, response["body"]["breakpoints"])
+        breakpoint_locations = response.body.breakpoints
+        for bp in expected_locations:
+            self.assertIn(bp, breakpoint_locations)
+
+        session.continue_to_exit()
diff --git a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_logpoints.py 
b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_logpoints.py
index d633f0b9ffde4..71c7c0cf9097c 100644
--- a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_logpoints.py
+++ b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_logpoints.py
@@ -2,109 +2,95 @@
 Test lldb-dap logpoints feature.
 """
 
-
-import dap_server
-import shutil
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
 import os
 
+from lldbsuite.test.decorators import (
+    skipIfTargetDoesNotSupportSharedLibraries,
+    skipIfWindows,
+)
+from lldbsuite.test.lldbtest import line_number
+from lldbsuite.test.tools.lldb_dap.dap_types import (
+    LaunchArgs,
+    SourceBreakpoint,
+    StoppedEvent,
+)
+from lldbsuite.test.tools.lldb_dap.lldb_dap_testcase import DAPTestCaseBase
+from lldbsuite.test.tools.lldb_dap.session_helpers import DAPTestSession
+
 
 @skipIfTargetDoesNotSupportSharedLibraries()
-class TestDAP_logpoints(lldbdap_testcase.DAPTestCaseBase):
+class TestDAP_logpoints(DAPTestCaseBase):
     def setUp(self):
-        lldbdap_testcase.DAPTestCaseBase.setUp(self)
+        DAPTestCaseBase.setUp(self)
 
         self.main_basename = "main-copy.cpp"
         self.main_path = 
os.path.realpath(self.getBuildArtifact(self.main_basename))
 
-    @skipIfWindows
-    def test_logmessage_basic(self):
-        """Tests breakpoint logmessage basic functionality."""
+    def stop_at_before_loop_line(self, session: DAPTestSession) -> 
StoppedEvent:
+        """Launch, set a breakpoint at 'before loop' line and stop there"""
         before_loop_line = line_number("main.cpp", "// before loop")
-        loop_line = line_number("main.cpp", "// break loop")
-        after_loop_line = line_number("main.cpp", "// after loop")
-
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-
-        # Set a breakpoint at a line before loop
-        before_loop_breakpoint_ids = self.set_source_breakpoints(
-            self.main_path, [before_loop_line]
-        )
-        self.assertEqual(len(before_loop_breakpoint_ids), 1, "expect one 
breakpoint")
-
-        self.dap_server.request_continue()
+        with session.configure(LaunchArgs(program)) as ctx:
+            [bp] = session.resolve_source_breakpoints(
+                self.main_path, [before_loop_line]
+            )
 
-        # Verify we hit the breakpoint before loop line
-        self.verify_breakpoint_hit(before_loop_breakpoint_ids)
+        return session.verify_stopped_on_breakpoint(bp, 
after=ctx.process_event)
 
-        # Swallow old console output
-        self.get_console()
+    @skipIfWindows
+    def test_logMessage_basic(self):
+        """Tests breakpoint logMessage basic functionality."""
+        session = self.build_and_create_session()
+        initial_stop = self.stop_at_before_loop_line(session)
+        source = self.getSourcePath("main.cpp")
+        loop_line = line_number(source, "// break loop")
+        after_loop_line = line_number(source, "// after loop")
 
         # Set two breakpoints:
-        # 1. First at the loop line with logMessage
-        # 2. Second guard breakpoint at a line after loop
+        # 1. First at the loop line with logMessage.
+        # 2. Second guard breakpoint at a line after loop.
         logMessage_prefix = "This is log message for { -- "
         logMessage = logMessage_prefix + "{i + 3}, {message}"
-        [loop_breakpoint_id, post_loop_breakpoint_id] = 
self.set_source_breakpoints(
+        [_, post_loop_breakpoint_id] = session.resolve_source_breakpoints(
             self.main_path,
-            [loop_line, after_loop_line],
-            [{"logMessage": logMessage}, {}],
+            [
+                SourceBreakpoint(loop_line, logMessage=logMessage),
+                SourceBreakpoint(after_loop_line),
+            ],
         )
 
-        # Continue to trigger the breakpoint with log messages
-        self.dap_server.request_continue()
+        # Continue and verify we hit the breakpoint after loop line.
+        post_loop_stop = 
session.continue_to_breakpoint(post_loop_breakpoint_id)
 
-        # Verify we hit the breakpoint after loop line
-        self.verify_breakpoint_hit([post_loop_breakpoint_id])
-
-        output = self.get_console()
-        lines = output.splitlines()
-        logMessage_output = []
-        for line in lines:
-            if line.startswith(logMessage_prefix):
-                logMessage_output.append(line)
-
-        # Verify logMessage count
-        loop_count = 10
-        self.assertEqual(len(logMessage_output), loop_count)
+        captured = session.collect_console(after=initial_stop, 
until=post_loop_stop)
+        logMessage_output = [
+            line
+            for line in captured.seen_texts.splitlines()
+            if line.startswith(logMessage_prefix)
+        ]
+        # Verify logMessage count.
+        self.assertEqual(len(logMessage_output), 10)
 
         message_addr_pattern = r"\b0x[0-9A-Fa-f]+\b"
         message_content = '"Hello from main!"'
-        # Verify log message match
+
+        # Verify logMessage match.
         for idx, logMessage_line in enumerate(logMessage_output):
             result = idx + 3
-            reg_str = (
-                f"{logMessage_prefix}{result}, {message_addr_pattern} 
{message_content}"
+            self.assertRegex(
+                logMessage_line,
+                f"{logMessage_prefix}{result}, {message_addr_pattern} 
{message_content}",
             )
-            self.assertRegex(logMessage_line, reg_str)
+        session.continue_to_exit()
 
     @skipIfWindows
     def test_logmessage_advanced(self):
         """Tests breakpoint logmessage functionality for complex expression."""
-        before_loop_line = line_number("main.cpp", "// before loop")
-        loop_line = line_number("main.cpp", "// break loop")
-        after_loop_line = line_number("main.cpp", "// after loop")
-
-        program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-
-        # Set a breakpoint at a line before loop
-        before_loop_breakpoint_ids = self.set_source_breakpoints(
-            self.main_path, [before_loop_line]
-        )
-        self.assertEqual(len(before_loop_breakpoint_ids), 1, "expect one 
breakpoint")
-
-        self.dap_server.request_continue()
-
-        # Verify we hit the breakpoint before loop line
-        self.verify_breakpoint_hit(before_loop_breakpoint_ids)
-
-        # Swallow old console output
-        self.get_console()
+        session = self.build_and_create_session()
+        initial_stop = self.stop_at_before_loop_line(session)
+        source = self.getSourcePath("main.cpp")
+        before_loop_line = line_number(source, "// break loop")
+        after_loop_line = line_number(source, "// after loop")
 
         # Set two breakpoints:
         # 1. First at the loop line with logMessage
@@ -114,91 +100,64 @@ def test_logmessage_advanced(self):
             logMessage_prefix
             + "{int y = 0; if (i % 3 == 0) { y = i + 3;} else {y = i * 3;} y}"
         )
-        [loop_breakpoint_id, post_loop_breakpoint_id] = 
self.set_source_breakpoints(
+        [_, post_loop_breakpoint_id] = session.resolve_source_breakpoints(
             self.main_path,
-            [loop_line, after_loop_line],
-            [{"logMessage": logMessage}, {}],
+            [
+                SourceBreakpoint(before_loop_line, logMessage=logMessage),
+                SourceBreakpoint(after_loop_line),
+            ],
         )
 
-        # Continue to trigger the breakpoint with log messages
-        self.dap_server.request_continue()
-
-        # Verify we hit the breakpoint after loop line
-        self.verify_breakpoint_hit([post_loop_breakpoint_id])
-
-        output = self.get_console()
-        lines = output.splitlines()
-        logMessage_output = []
-        for line in lines:
-            if line.startswith(logMessage_prefix):
-                logMessage_output.append(line)
-
-        # Verify logMessage count
-        loop_count = 10
-        self.assertEqual(len(logMessage_output), loop_count)
-
-        # Verify log message match
+        post_loop_stop = 
session.continue_to_breakpoint(post_loop_breakpoint_id)
+        captured = session.collect_console(after=initial_stop, 
until=post_loop_stop)
+        logMessage_output = [
+            line
+            for line in captured.seen_texts.splitlines()
+            if line.startswith(logMessage_prefix)
+        ]
+        # Verify logMessage count.
+        self.assertEqual(len(logMessage_output), 10)
+
+        # Verify logMessage match.
         for idx, logMessage_line in enumerate(logMessage_output):
             result = idx + 3 if idx % 3 == 0 else idx * 3
             self.assertEqual(logMessage_line, logMessage_prefix + str(result))
 
     @skipIfWindows
     def test_logmessage_format(self):
-        """
-        Tests breakpoint logmessage functionality with format.
-        """
-        before_loop_line = line_number("main.cpp", "// before loop")
-        loop_line = line_number("main.cpp", "// break loop")
-        after_loop_line = line_number("main.cpp", "// after loop")
-
-        program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-
-        # Set a breakpoint at a line before loop
-        before_loop_breakpoint_ids = self.set_source_breakpoints(
-            self.main_path, [before_loop_line]
-        )
-        self.assertEqual(len(before_loop_breakpoint_ids), 1, "expect one 
breakpoint")
-
-        self.dap_server.request_continue()
-
-        # Verify we hit the breakpoint before loop line
-        self.verify_breakpoint_hit(before_loop_breakpoint_ids)
-
-        # Swallow old console output
-        self.get_console()
+        """Tests breakpoint logmessage functionality with format."""
+        session = self.build_and_create_session()
+        initial_stop = self.stop_at_before_loop_line(session)
+        source = self.getSourcePath("main.cpp")
+        loop_line = line_number(source, "// break loop")
+        after_loop_line = line_number(source, "// after loop")
 
         # Set two breakpoints:
-        # 1. First at the loop line with logMessage
-        # 2. Second guard breakpoint at a line after loop
+        # 1. First at the loop line with logMessage.
+        # 2. Second guard breakpoint at a line after loop.
         logMessage_prefix = "This is log message for -- "
         logMessage_with_format = "part1\tpart2\bpart3\x64part4"
         logMessage_with_format_raw = r"part1\tpart2\bpart3\x64part4"
         logMessage = logMessage_prefix + logMessage_with_format_raw + "{i - 1}"
-        [loop_breakpoint_id, post_loop_breakpoint_id] = 
self.set_source_breakpoints(
+        [_, post_loop_breakpoint_id] = session.resolve_source_breakpoints(
             self.main_path,
-            [loop_line, after_loop_line],
-            [{"logMessage": logMessage}, {}],
+            [
+                SourceBreakpoint(loop_line, logMessage=logMessage),
+                SourceBreakpoint(after_loop_line),
+            ],
         )
 
-        # Continue to trigger the breakpoint with log messages
-        self.dap_server.request_continue()
-
-        # Verify we hit the breakpoint after loop line
-        self.verify_breakpoint_hit([post_loop_breakpoint_id])
-
-        output = self.get_console()
-        lines = output.splitlines()
-        logMessage_output = []
-        for line in lines:
-            if line.startswith(logMessage_prefix):
-                logMessage_output.append(line)
-
-        # Verify logMessage count
-        loop_count = 10
-        self.assertEqual(len(logMessage_output), loop_count)
-
-        # Verify log message match
+        post_loop_stop = 
session.continue_to_breakpoint(post_loop_breakpoint_id)
+        captured = session.collect_console(after=initial_stop, 
until=post_loop_stop)
+        logMessage_output = [
+            line
+            for line in captured.seen_texts.splitlines()
+            if line.startswith(logMessage_prefix)
+        ]
+        # Verify logMessage count.
+        self.assertEqual(len(logMessage_output), 10)
+
+        # Verify logMessage match.
         for idx, logMessage_line in enumerate(logMessage_output):
             result = idx - 1
             self.assertEqual(
@@ -208,64 +167,41 @@ def test_logmessage_format(self):
 
     @skipIfWindows
     def test_logmessage_format_failure(self):
-        """
-        Tests breakpoint logmessage format with parsing failure.
-        """
-        before_loop_line = line_number("main.cpp", "// before loop")
-        loop_line = line_number("main.cpp", "// break loop")
-        after_loop_line = line_number("main.cpp", "// after loop")
-
-        program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-
-        # Set a breakpoint at a line before loop
-        before_loop_breakpoint_ids = self.set_source_breakpoints(
-            self.main_path, [before_loop_line]
-        )
-        self.assertEqual(len(before_loop_breakpoint_ids), 1, "expect one 
breakpoint")
-
-        self.dap_server.request_continue()
-
-        # Verify we hit the breakpoint before loop line
-        self.verify_breakpoint_hit(before_loop_breakpoint_ids)
-
-        # Swallow old console output
-        self.get_console()
+        """Tests breakpoint logmessage format with parsing failure."""
+        session = self.build_and_create_session()
+        initial_stop = self.stop_at_before_loop_line(session)
+        source = self.getSourcePath("main.cpp")
+        loop_line = line_number(source, "// break loop")
+        after_loop_line = line_number(source, "// after loop")
 
         # Set two breakpoints:
-        # 1. First at the loop line with logMessage
-        # 2. Second guard breakpoint at a line after loop
+        # 1. First at the loop line with logMessage.
+        # 2. Second guard breakpoint at a line after loop.
         logMessage_prefix = "This is log message for -- "
         # log message missing hex number.
-        logMessage_with_format_raw = r"part1\x"
-        logMessage = logMessage_prefix + logMessage_with_format_raw
-        [loop_breakpoint_id, post_loop_breakpoint_id] = 
self.set_source_breakpoints(
+        logMessage = logMessage_prefix + r"part1\x"
+        [loop_breakpoint_id, _] = session.resolve_source_breakpoints(
             self.main_path,
-            [loop_line, after_loop_line],
-            [{"logMessage": logMessage}, {}],
+            [
+                SourceBreakpoint(loop_line, logMessage=logMessage),
+                SourceBreakpoint(after_loop_line),
+            ],
         )
 
-        # Continue to trigger the breakpoint with log messages
-        self.dap_server.request_continue()
-
-        # Verify we hit logpoint breakpoint if it's format has error.
-        self.verify_breakpoint_hit([loop_breakpoint_id])
-
-        output = self.get_console()
-        lines = output.splitlines()
+        # The adapter emits the format error to the console during
+        # setBreakpoints and falls back to a real stop at the loop breakpoint
+        # when execution resumes.
+        loop_stop_event = session.continue_to_breakpoint(loop_breakpoint_id)
+        captured = session.collect_console(after=initial_stop, 
until=loop_stop_event)
 
         failure_prefix = "Log message has error:"
-        logMessage_output = []
-        logMessage_failure_output = []
-        for line in lines:
-            if line.startswith(logMessage_prefix):
-                logMessage_output.append(line)
-            elif line.startswith(failure_prefix):
-                logMessage_failure_output.append(line)
-
-        # Verify logMessage failure message
+        logMessage_failure_output = [
+            line.strip()
+            for line in captured.seen_texts.splitlines()
+            if line.startswith(failure_prefix)
+        ]
         self.assertEqual(len(logMessage_failure_output), 1)
         self.assertEqual(
-            logMessage_failure_output[0].strip(),
+            logMessage_failure_output[0],
             failure_prefix + " missing hex number following '\\x'",
         )
diff --git a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setBreakpoints.py 
b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setBreakpoints.py
index 2c0b9c5f6e07f..66281576563f3 100644
--- a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setBreakpoints.py
+++ b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setBreakpoints.py
@@ -2,26 +2,32 @@
 Test lldb-dap setBreakpoints request
 """
 
-from dap_server import Source
-import shutil
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
 import os
+import shutil
+from typing import Dict
+
+from lldbsuite.test.decorators import skipIfWasm, skipIfWindows
+from lldbsuite.test.lldbtest import line_number
+from lldbsuite.test.tools.lldb_dap.dap_types import (
+    DAPTestGetTargetBreakpointsArgs,
+    LaunchArgs,
+    SetBreakpointsArgs,
+    Source,
+    SourceBreakpoint,
+)
+from lldbsuite.test.tools.lldb_dap.lldb_dap_testcase import DAPTestCaseBase
 
 
 @skipIfWasm  # inferior built without exception support
-class TestDAP_setBreakpoints(lldbdap_testcase.DAPTestCaseBase):
+class TestDAP_setBreakpoints(DAPTestCaseBase):
     SHARED_BUILD_TESTCASE = False
 
     def setUp(self):
-        lldbdap_testcase.DAPTestCaseBase.setUp(self)
+        DAPTestCaseBase.setUp(self)
 
         self.main_basename = "main-copy.cpp"
         self.main_path = 
os.path.realpath(self.getBuildArtifact(self.main_basename))
 
-    @skipIfTargetDoesNotSupportSharedLibraries()
     @skipIfWindows
     def test_source_map(self):
         """
@@ -30,8 +36,7 @@ def test_source_map(self):
         with the corresponding source maps to have breakpoints and frames
         working.
         """
-        self.build_and_create_debug_adapter()
-
+        session = self.build_and_create_session()
         other_basename = "other-copy.c"
         other_path = self.getBuildArtifact(other_basename)
 
@@ -43,7 +48,7 @@ def test_source_map(self):
         new_main_path = os.path.join(new_main_folder, self.main_basename)
         new_other_path = os.path.join(new_other_folder, other_basename)
 
-        # move the sources
+        # Move the sources.
         os.mkdir(new_main_folder)
         os.mkdir(new_other_folder)
         shutil.move(self.main_path, new_main_path)
@@ -54,57 +59,59 @@ def test_source_map(self):
 
         program = self.getBuildArtifact("a.out")
         source_map = [
-            [source_folder, new_main_folder],
-            [source_folder, new_other_folder],
+            (source_folder, new_main_folder),
+            (source_folder, new_other_folder),
         ]
-        self.launch(program, sourceMap=source_map)
-
-        # breakpoint in main.cpp
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=new_main_path), [main_line]
+        with session.configure(LaunchArgs(program, sourceMap=source_map)) as 
ctx:
+            # Breakpoint in main.cpp.
+            response = session.set_source_breakpoints(new_main_path, 
[main_line])
+            self.assertEqual(len(response.body.breakpoints), 1)
+            [main_bp] = response.body.breakpoints
+            self.assertEqual(main_bp.line, main_line)
+            self.assertTrue(main_bp.verified)
+            breakpoint_source = self.expect_not_none(main_bp.source)
+            self.assertEqual(self.main_basename, breakpoint_source.name)
+            self.assertEqual(new_main_path, breakpoint_source.path)
+
+            # 2nd breakpoint, which is from a dynamically loaded library.
+            response = session.set_source_breakpoints(new_other_path, 
[other_line])
+            [other_bp] = response.body.breakpoints
+            self.assertEqual(other_bp.line, other_line)
+            self.assertFalse(other_bp.verified)
+            breakpoint_source = self.expect_not_none(other_bp.source)
+            self.assertEqual(other_basename, breakpoint_source.name)
+            self.assertEqual(new_other_path, breakpoint_source.path)
+            other_bp_id = self.expect_not_none(other_bp.id)
+
+        stop_event = session.verify_stopped_on_breakpoint(
+            other_bp_id, after=ctx.process_event
         )
-        breakpoints = response["body"]["breakpoints"]
-        self.assertEqual(len(breakpoints), 1)
-        breakpoint = breakpoints[0]
-        self.assertEqual(breakpoint["line"], main_line)
-        self.assertTrue(breakpoint["verified"])
-        self.assertEqual(self.main_basename, breakpoint["source"]["name"])
-        self.assertEqual(new_main_path, breakpoint["source"]["path"])
-
-        # 2nd breakpoint, which is from a dynamically loaded library
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=new_other_path), [other_line]
-        )
-        breakpoints = response["body"]["breakpoints"]
-        breakpoint = breakpoints[0]
-        self.assertEqual(breakpoint["line"], other_line)
-        self.assertFalse(breakpoint["verified"])
-        self.assertEqual(other_basename, breakpoint["source"]["name"])
-        self.assertEqual(new_other_path, breakpoint["source"]["path"])
-        other_breakpoint_id = breakpoint["id"]
-
-        self.dap_server.request_continue()
-        self.verify_breakpoint_hit([other_breakpoint_id])
-
-        # 2nd breakpoint again, which should be valid at this point
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=new_other_path), [other_line]
-        )
-        breakpoints = response["body"]["breakpoints"]
-        breakpoint = breakpoints[0]
-        self.assertEqual(breakpoint["line"], other_line)
-        self.assertTrue(breakpoint["verified"])
-        self.assertEqual(other_basename, breakpoint["source"]["name"])
-        self.assertEqual(new_other_path, breakpoint["source"]["path"])
 
-        # now we check the stack trace making sure that we got mapped source 
paths
-        frames = self.dap_server.request_stackTrace()["body"]["stackFrames"]
+        # 2nd breakpoint again, which should be valid at this point.
+        response = session.set_source_breakpoints(new_other_path, [other_line])
+        [other_bp] = response.body.breakpoints
+        self.assertEqual(other_bp.line, other_line)
+        self.assertTrue(other_bp.verified)
+        breakpoint_source = self.expect_not_none(other_bp.source)
+        self.assertEqual(other_basename, breakpoint_source.name)
+        self.assertEqual(new_other_path, breakpoint_source.path)
+
+        # Now we check the stack trace making sure that we got mapped source 
paths.
+        frame_ctxs = session.thread_context_from(stop_event).frames()
+        frames = [ctx.frame for ctx in frame_ctxs]
+
+        frame0_source = self.expect_not_none(frames[0].source)
+        self.assertEqual(frame0_source.name, other_basename)
+        self.assertEqual(frame0_source.path, new_other_path)
 
-        self.assertEqual(frames[0]["source"]["name"], other_basename)
-        self.assertEqual(frames[0]["source"]["path"], new_other_path)
+        frame1_source = self.expect_not_none(frames[1].source)
+        self.assertEqual(frame1_source.name, self.main_basename)
+        self.assertEqual(frame1_source.path, new_main_path)
 
-        self.assertEqual(frames[1]["source"]["name"], self.main_basename)
-        self.assertEqual(frames[1]["source"]["path"], new_main_path)
+        # Clear all breakpoints.
+        session.set_source_breakpoints(new_main_path, [])
+        session.set_source_breakpoints(new_other_path, [])
+        session.continue_to_exit()
 
     @skipIfWindows
     def test_set_and_clear(self):
@@ -129,25 +136,25 @@ def test_set_and_clear(self):
         # without launching or attaching to a process, so we must start a
         # process in order to be able to set breakpoints.
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
+        session = self.build_and_create_session()
+        session.initialize_and_launch(LaunchArgs(program))
 
-        # Set 3 breakpoints and verify that they got set correctly
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path), lines
-        )
-        line_to_id = {}
-        breakpoints = response["body"]["breakpoints"]
+        # Set 3 breakpoints and verify that they got set correctly.
+        response = session.set_source_breakpoints(self.main_path, lines)
+        breakpoints = response.body.breakpoints
         self.assertEqual(
             len(breakpoints),
             len(lines),
-            "expect %u source breakpoints" % (len(lines)),
+            f"expect {len(lines)} source breakpoints",
         )
+        line_to_id: Dict[int, int] = {}
         for index, breakpoint in enumerate(breakpoints):
-            line = breakpoint["line"]
+            line = self.expect_not_none(breakpoint.line)
             self.assertEqual(line, lines[index])
-            # Store the "id" of the breakpoint that was set for later
-            line_to_id[line] = breakpoint["id"]
-            self.assertTrue(breakpoint["verified"], "expect breakpoint 
verified")
+            # Store the "id" of the breakpoint that was set for later.
+            breakpoint_id = self.expect_not_none(breakpoint.id)
+            line_to_id[line] = breakpoint_id
+            self.assertTrue(breakpoint.verified, "expect breakpoint verified")
 
         # There is no breakpoint delete packet, clients just send another
         # setBreakpoints packet with the same source file with fewer lines.
@@ -159,110 +166,96 @@ def test_set_and_clear(self):
         lines.remove(second_line)
         # Set 2 breakpoints and verify that the previous breakpoints that were
         # set above are still set.
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path), lines
-        )
-        breakpoints = response["body"]["breakpoints"]
+        response = session.set_source_breakpoints(self.main_path, lines)
+        breakpoints = response.body.breakpoints
         self.assertEqual(
-            len(breakpoints),
-            len(lines),
-            "expect %u source breakpoints" % (len(lines)),
+            len(breakpoints), len(lines), f"expect {len(lines)} source 
breakpoints"
         )
-        for index, breakpoint in enumerate(breakpoints):
-            line = breakpoint["line"]
-            self.assertEqual(line, lines[index])
+        for expected_line, breakpoint in zip(lines, breakpoints):
+            line = self.expect_not_none(breakpoint.line)
+            self.assertEqual(line, expected_line)
             # Verify the same breakpoints are still set within LLDB by
-            # making sure the breakpoint ID didn't change
+            # making sure the breakpoint ID didn't change.
             self.assertEqual(
                 line_to_id[line],
-                breakpoint["id"],
+                breakpoint.id,
                 "verify previous breakpoints stayed the same",
             )
-            self.assertTrue(breakpoint["verified"], "expect breakpoint still 
verified")
+            self.assertTrue(breakpoint.verified, "expect breakpoint still 
verified")
 
         # Now get the full list of breakpoints set in the target and verify
         # we have only 2 breakpoints set. The response above could have told
         # us about 2 breakpoints, but we want to make sure we don't have the
         # third one still set in the target
-        response = self.dap_server.request_testGetTargetBreakpoints()
-        breakpoints = response["body"]["breakpoints"]
+        response = 
session.send_request(DAPTestGetTargetBreakpointsArgs()).result()
+        breakpoints = response.body.breakpoints
         self.assertEqual(
             len(breakpoints),
             len(lines),
-            "expect %u source breakpoints" % (len(lines)),
+            f"expect {len(lines)} source breakpoints",
         )
         for breakpoint in breakpoints:
-            line = breakpoint["line"]
+            line = self.expect_not_none(breakpoint.line)
             # Verify the same breakpoints are still set within LLDB by
             # making sure the breakpoint ID didn't change
             self.assertEqual(
                 line_to_id[line],
-                breakpoint["id"],
+                breakpoint.id,
                 "verify previous breakpoints stayed the same",
             )
             self.assertIn(line, lines, "line expected in lines array")
-            self.assertTrue(breakpoint["verified"], "expect breakpoint still 
verified")
+            self.assertTrue(breakpoint.verified, "expect breakpoint still 
verified")
 
         # Now clear all breakpoints for the source file by passing down an
-        # empty lines array
+        # empty lines array.
         lines = []
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path), lines
-        )
-        breakpoints = response["body"]["breakpoints"]
+        response = session.set_source_breakpoints(self.main_path, lines)
+        breakpoints = response.body.breakpoints
         self.assertEqual(
             len(breakpoints),
             len(lines),
-            "expect %u source breakpoints" % (len(lines)),
+            f"expect {len(lines)} source breakpoints",
         )
 
         # Verify with the target that all breakpoints have been cleared
-        response = self.dap_server.request_testGetTargetBreakpoints()
-        breakpoints = response["body"]["breakpoints"]
+        response = 
session.send_request(DAPTestGetTargetBreakpointsArgs()).result()
+        breakpoints = response.body.breakpoints
         self.assertEqual(
             len(breakpoints),
             len(lines),
-            "expect %u source breakpoints" % (len(lines)),
+            f"expect {len(lines)} source breakpoints",
         )
 
         # Now set a breakpoint again in the same source file and verify it
         # was added.
         lines = [second_line]
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path), lines
+        response = session.set_source_breakpoints(self.main_path, lines)
+        breakpoints = response.body.breakpoints
+        self.assertEqual(
+            len(breakpoints),
+            len(lines),
+            f"expect {len(lines)} source breakpoints",
         )
-        if response:
-            breakpoints = response["body"]["breakpoints"]
-            self.assertEqual(
-                len(breakpoints),
-                len(lines),
-                "expect %u source breakpoints" % (len(lines)),
-            )
-            for breakpoint in breakpoints:
-                line = breakpoint["line"]
-                self.assertIn(line, lines, "line expected in lines array")
-                self.assertTrue(
-                    breakpoint["verified"], "expect breakpoint still verified"
-                )
+        for breakpoint in breakpoints:
+            line = self.expect_not_none(breakpoint.line)
+            self.assertIn(line, lines, "line expected in lines array")
+            self.assertTrue(breakpoint.verified, "expect breakpoint still 
verified")
 
         # Now get the full list of breakpoints set in the target and verify
         # we have only 2 breakpoints set. The response above could have told
         # us about 2 breakpoints, but we want to make sure we don't have the
-        # third one still set in the target
-        response = self.dap_server.request_testGetTargetBreakpoints()
-        if response:
-            breakpoints = response["body"]["breakpoints"]
-            self.assertEqual(
-                len(breakpoints),
-                len(lines),
-                "expect %u source breakpoints" % (len(lines)),
-            )
-            for breakpoint in breakpoints:
-                line = breakpoint["line"]
-                self.assertIn(line, lines, "line expected in lines array")
-                self.assertTrue(
-                    breakpoint["verified"], "expect breakpoint still verified"
-                )
+        # third one still set in the target.
+        response = 
session.send_request(DAPTestGetTargetBreakpointsArgs()).result()
+        breakpoints = response.body.breakpoints
+        self.assertEqual(
+            len(breakpoints),
+            len(lines),
+            f"expect {len(lines)} source breakpoints",
+        )
+        for breakpoint in breakpoints:
+            line = self.expect_not_none(breakpoint.line)
+            self.assertIn(line, lines, "line expected in lines array")
+            self.assertTrue(breakpoint.verified, "expect breakpoint still 
verified")
 
     @skipIfWindows
     def test_clear_breakpoints_unset_breakpoints(self):
@@ -278,37 +271,33 @@ def test_clear_breakpoints_unset_breakpoints(self):
         # without launching or attaching to a process, so we must start a
         # process in order to be able to set breakpoints.
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-
-        # Set one breakpoint and verify that it got set correctly.
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path), lines
-        )
-        line_to_id = {}
-        breakpoints = response["body"]["breakpoints"]
-        self.assertEqual(
-            len(breakpoints), len(lines), "expect %u source breakpoints" % 
(len(lines))
-        )
-        for index, breakpoint in enumerate(breakpoints):
-            line = breakpoint["line"]
-            self.assertEqual(line, lines[index])
-            # Store the "id" of the breakpoint that was set for later
-            line_to_id[line] = breakpoint["id"]
-            self.assertTrue(breakpoint["verified"], "expect breakpoint 
verified")
-
-        # Now clear all breakpoints for the source file by not setting the
-        # lines array.
-        lines = None
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path), lines
-        )
-        breakpoints = response["body"]["breakpoints"]
-        self.assertEqual(len(breakpoints), 0, "expect no source breakpoints")
-
-        # Verify with the target that all breakpoints have been cleared.
-        response = self.dap_server.request_testGetTargetBreakpoints()
-        breakpoints = response["body"]["breakpoints"]
-        self.assertEqual(len(breakpoints), 0, "expect no source breakpoints")
+        session = self.build_and_create_session()
+        with session.configure(LaunchArgs(program)):
+            # Set one breakpoint and verify that it got set correctly.
+            response = session.set_source_breakpoints(self.main_path, lines)
+            breakpoints = response.body.breakpoints
+            self.assertEqual(
+                len(breakpoints), len(lines), f"expect {len(lines)} source 
breakpoints"
+            )
+            for expected_line, breakpoint in zip(lines, breakpoints):
+                line = self.expect_not_none(breakpoint.line)
+                self.assertEqual(line, expected_line)
+                self.expect_not_none(breakpoint.id)
+                self.assertTrue(breakpoint.verified, "expect breakpoint 
verified")
+
+            # Now clear all breakpoints for the source file by not setting the
+            # lines or breakpoints array.
+            set_bp_args = 
SetBreakpointsArgs(Source.create(path=self.main_path))
+            response = session.send_request(set_bp_args).result()
+            breakpoints = response.body.breakpoints
+            self.assertEqual(len(breakpoints), 0, "expect no source 
breakpoints")
+
+            # Verify with the target that all breakpoints have been cleared.
+            response = 
session.send_request(DAPTestGetTargetBreakpointsArgs()).result()
+            breakpoints = response.body.breakpoints
+            self.assertEqual(len(breakpoints), 0, "expect no source 
breakpoints")
+
+        session.verify_process_exited()
 
     @skipIfWindows
     def test_functionality(self):
@@ -317,122 +306,136 @@ def test_functionality(self):
         loop_line = line_number("main.cpp", "// break loop")
 
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-        # Set a breakpoint at the loop line with no condition and no
-        # hitCondition
-        breakpoint_ids = self.set_source_breakpoints(self.main_path, 
[loop_line])
-        self.assertEqual(len(breakpoint_ids), 1, "expect one breakpoint")
-        self.dap_server.request_continue()
-
-        # Verify we hit the breakpoint we just set
-        self.verify_breakpoint_hit(breakpoint_ids)
-
-        # Make sure i is zero at first breakpoint
-        i = int(self.dap_server.get_local_variable_value("i"))
-        self.assertEqual(i, 0, "i != 0 after hitting breakpoint")
-
-        # Update the condition on our breakpoint
-        new_breakpoint_ids = self.set_source_breakpoints(
-            self.main_path, [loop_line], [{"condition": "i==4"}]
+
+        session = self.build_and_create_session()
+        with session.configure(LaunchArgs(program)) as ctx:
+            # Set a breakpoint at the loop line with no condition and no
+            # hitCondition.
+            [loop_bp] = session.resolve_source_breakpoints(self.main_path, 
[loop_line])
+        process_event = ctx.process_event
+
+        # Verify we hit the breakpoint we just set.
+        stop_event = session.verify_stopped_on_breakpoint(loop_bp, 
after=process_event)
+
+        # Make sure i is zero at first breakpoint.
+        thread_ctx = session.thread_context_from(stop_event)
+        i_var = thread_ctx.top_frame().locals["i"]
+        self.assertEqual(i_var.value_as_int, 0, "i != 0 after hitting 
breakpoint")
+
+        # Update the condition on our breakpoint.
+        [condition_bp] = session.resolve_source_breakpoints(
+            self.main_path, [SourceBreakpoint(loop_line, condition="i==4")]
         )
         self.assertEqual(
-            breakpoint_ids,
-            new_breakpoint_ids,
-            "existing breakpoint should have its condition " "updated",
+            loop_bp,
+            condition_bp,
+            "existing breakpoint should have its condition updated",
         )
 
-        self.continue_to_breakpoints(breakpoint_ids)
-        i = int(self.dap_server.get_local_variable_value("i"))
-        self.assertEqual(i, 4, "i != 4 showing conditional works")
+        stop_event = session.continue_to_breakpoint(loop_bp)
+        thread_ctx = session.thread_context_from(stop_event)
+        i_var = thread_ctx.top_frame().locals["i"]
+        self.assertEqual(i_var.value_as_int, 4, "i != 4 showing conditional 
works")
 
-        new_breakpoint_ids = self.set_source_breakpoints(
-            self.main_path, [loop_line], [{"hitCondition": "2"}]
+        # Update the hitCondition on our breakpoint.
+        [hit_condition_bp] = session.resolve_source_breakpoints(
+            self.main_path, [SourceBreakpoint(loop_line, hitCondition="2")]
         )
-
         self.assertEqual(
-            breakpoint_ids,
-            new_breakpoint_ids,
-            "existing breakpoint should have its condition " "updated",
+            loop_bp,
+            hit_condition_bp,
+            "existing breakpoint should have its condition updated",
         )
 
-        # Continue with a hitCondition of 2 and expect it to skip 1 value
-        self.continue_to_breakpoints(breakpoint_ids)
-        i = int(self.dap_server.get_local_variable_value("i"))
-        self.assertEqual(i, 6, "i != 6 showing hitCondition works")
+        # Continue with a hitCondition of 2 and expect it to skip 1 value.
+        stop_event = session.continue_to_breakpoint(loop_bp)
+        i_var = thread_ctx.top_frame().locals["i"]
+        self.assertEqual(i_var.value_as_int, 6, "i != 6 showing hitCondition 
works")
 
-        # continue after hitting our hitCondition and make sure it only goes
-        # up by 1
-        self.continue_to_breakpoints(breakpoint_ids)
-        i = int(self.dap_server.get_local_variable_value("i"))
-        self.assertEqual(i, 7, "i != 7 showing post hitCondition hits every 
time")
+        # Continue after hitting our hitCondition and make sure it only goes
+        # up by 1.
+        stop_event = session.continue_to_breakpoint(loop_bp)
+        i_var = thread_ctx.top_frame().locals["i"]
+        self.assertEqual(
+            i_var.value_as_int, 7, "i != 7 showing post hitCondition hits 
every time"
+        )
+
+        # Clear breakpoints and exit.
+        session.set_source_breakpoints(self.main_path, [])
+        session.continue_to_exit()
 
     @skipIfWindows
     def test_column_breakpoints(self):
         """Test setting multiple breakpoints in the same line at different 
columns."""
+        session = self.build_and_create_session()
         loop_line = line_number("main.cpp", "// break loop")
 
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-
-        # Set two breakpoints on the loop line at different columns.
-        columns = [13, 39]
-        response = self.dap_server.request_setBreakpoints(
-            Source.build(path=self.main_path),
-            [loop_line, loop_line],
-            list({"column": c} for c in columns),
-        )
-
-        # Verify the breakpoints were set correctly
-        breakpoints = response["body"]["breakpoints"]
-        breakpoint_ids = []
-        self.assertEqual(
-            len(breakpoints),
-            len(columns),
-            "expect %u source breakpoints" % (len(columns)),
-        )
-        for index, breakpoint in enumerate(breakpoints):
-            self.assertEqual(breakpoint["line"], loop_line)
-            self.assertEqual(breakpoint["column"], columns[index])
-            self.assertTrue(breakpoint["verified"], "expect breakpoint 
verified")
-            breakpoint_ids.append(breakpoint["id"])
+        with session.configure(LaunchArgs(program)):
+            # Set two breakpoints on the loop line at different columns.
+            columns = [13, 39]
+            source_bps = [
+                SourceBreakpoint(line=loop_line, column=column) for column in 
columns
+            ]
+            response = session.set_source_breakpoints(self.main_path, 
source_bps)
+
+            # Verify the breakpoints were set correctly.
+            breakpoints = response.body.breakpoints
+            self.assertEqual(
+                len(breakpoints),
+                len(columns),
+                f"expect {len(columns)} source breakpoints",
+            )
+            breakpoint_ids: list[int] = []
+            for column, breakpoint in zip(columns, breakpoints):
+                self.assertEqual(breakpoint.line, loop_line)
+                self.assertEqual(breakpoint.column, column)
+                self.assertTrue(breakpoint.verified, "expect breakpoint 
verified")
+                breakpoint_id = self.expect_not_none(breakpoint.id)
+                breakpoint_ids.append(breakpoint_id)
 
         # Continue to the first breakpoint,
-        self.continue_to_breakpoints([breakpoint_ids[0]])
+        stop_event = session.verify_stopped_on_breakpoint(
+            breakpoint_ids[0], after=response
+        )
 
         # We should have stopped right before the call to `twelve`.
         # Step into and check we are inside `twelve`.
-        self.stepIn()
-        func_name = self.get_stackFrames()[0]["name"]
+        thread_ctx = session.thread_context_from(stop_event)
+        thread_ctx.step_in()
+        func_name = thread_ctx.top_frame().name
         self.assertEqual(func_name, "twelve(int)")
 
         # Continue to the second breakpoint.
-        self.continue_to_breakpoints([breakpoint_ids[1]])
+        stop_event = session.continue_to_breakpoint(breakpoint_ids[1])
 
         # We should have stopped right before the call to `fourteen`.
         # Step into and check we are inside `fourteen`.
-        self.stepIn()
-        func_name = self.get_stackFrames()[0]["name"]
+        thread_ctx.step_in()
+        func_name = thread_ctx.top_frame().name
         self.assertEqual(func_name, "a::fourteen(int)")
 
+        session.set_source_breakpoints(self.main_path, [])
+        session.continue_to_exit()
+
     @skipIfWindows
     def test_hit_multiple_breakpoints(self):
         """Test that if we hit multiple breakpoints at the same address, they
         all appear in the stop reason."""
+        session = self.build_and_create_session()
         breakpoint_lines = [
             line_number("main.cpp", "// break non-breakpointable line"),
             line_number("main.cpp", "// before loop"),
         ]
 
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
+        with session.configure(LaunchArgs(program)) as ctx:
+            # Set a pair of breakpoints that will both resolve to the same 
address.
+            breakpoint_ids = session.resolve_source_breakpoints(
+                self.main_path, breakpoint_lines
+            )
 
-        # Set a pair of breakpoints that will both resolve to the same address.
-        breakpoint_ids = [
-            int(bp_id)
-            for bp_id in self.set_source_breakpoints(self.main_path, 
breakpoint_lines)
-        ]
-        self.assertEqual(len(breakpoint_ids), 2, "expected two breakpoints")
-        self.dap_server.request_continue()
-        print(breakpoint_ids)
-        # Verify we hit both of the breakpoints we just set
-        self.verify_all_breakpoints_hit(breakpoint_ids)
+        # Verify we hit both of the breakpoints we just set.
+        session.verify_multiple_breakpoints_hit(breakpoint_ids, 
after=ctx.process_event)
+
+        session.continue_to_exit()
diff --git 
a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setExceptionBreakpoints.py 
b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setExceptionBreakpoints.py
index aec028b44e29a..aa28f7f8096de 100644
--- a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setExceptionBreakpoints.py
+++ b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setExceptionBreakpoints.py
@@ -2,13 +2,17 @@
 Test lldb-dap setExceptionBreakpoints request
 """
 
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-import lldbdap_testcase
+
+from lldbsuite.test.decorators import (
+    skipIfTargetDoesNotSupportSharedLibraries,
+    skipIfWindows,
+)
+from lldbsuite.test.tools.lldb_dap.dap_types import LaunchArgs
+from lldbsuite.test.tools.lldb_dap.lldb_dap_testcase import DAPTestCaseBase
 
 
 @skipIfTargetDoesNotSupportSharedLibraries()
-class TestDAP_setExceptionBreakpoints(lldbdap_testcase.DAPTestCaseBase):
+class TestDAP_setExceptionBreakpoints(DAPTestCaseBase):
     @skipIfWindows
     def test_functionality(self):
         """Tests setting and clearing exception breakpoints.
@@ -28,17 +32,22 @@ def test_functionality(self):
         # without launching or attaching to a process, so we must start a
         # process in order to be able to set breakpoints.
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
+        session = self.build_and_create_session()
 
-        response = self.dap_server.request_setExceptionBreakpoints(
-            filters=["cpp_throw", "cpp_catch"],
-        )
-        if response:
-            self.assertTrue(response["success"])
+        with session.configure(LaunchArgs(program)) as ctx:
+            response = session.set_exception_breakpoints(
+                filters=["cpp_throw", "cpp_catch"]
+            )
+            breakpoints = self.expect_not_none(response.body.breakpoints)
+            for bp in breakpoints:
+                self.assertTrue(bp.verified, True)
 
-        self.continue_to_exception_breakpoint(
-            expected_description=r"breakpoint 1\.1", expected_text=r"C\+\+ 
Throw"
+        session.verify_stopped_on_exception(
+            expected_description=r"breakpoint 1\.1",
+            expected_text=r"C\+\+ Throw",
+            after=ctx.process_event,
         )
-        self.continue_to_exception_breakpoint(
+        session.continue_to_exception_breakpoint(
             expected_description=r"breakpoint 2\.1", expected_text=r"C\+\+ 
Catch"
         )
+        session.continue_to_exit()
diff --git 
a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setFunctionBreakpoints.py 
b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setFunctionBreakpoints.py
index d99f18ccb2000..f121b80648228 100644
--- a/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setFunctionBreakpoints.py
+++ b/lldb/test/API/tools/lldb-dap/breakpoint/TestDAP_setFunctionBreakpoints.py
@@ -1,17 +1,21 @@
 """
-Test lldb-dap setBreakpoints request
+Test lldb-dap setFunctionBreakpoints request
 """
 
-
-import dap_server
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
+from lldbsuite.test.decorators import (
+    skipIfTargetDoesNotSupportSharedLibraries,
+    skipIfWindows,
+)
+from lldbsuite.test.tools.lldb_dap.dap_types import (
+    DAPTestGetTargetBreakpointsArgs,
+    FunctionBreakpoint,
+    LaunchArgs,
+)
+from lldbsuite.test.tools.lldb_dap.lldb_dap_testcase import DAPTestCaseBase
 
 
 @skipIfTargetDoesNotSupportSharedLibraries()
-class TestDAP_setFunctionBreakpoints(lldbdap_testcase.DAPTestCaseBase):
+class TestDAP_setFunctionBreakpoints(DAPTestCaseBase):
     @skipIfWindows
     def test_set_and_clear(self):
         """Tests setting and clearing function breakpoints.
@@ -29,145 +33,134 @@ def test_set_and_clear(self):
         # Visual Studio Code Debug Adapters have no way to specify the file
         # without launching or attaching to a process, so we must start a
         # process in order to be able to set breakpoints.
-        program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-        bp_id_12 = None
-        functions = ["twelve"]
-        # Set a function breakpoint at 'twelve'
-        response = self.dap_server.request_setFunctionBreakpoints(functions)
-        if response:
-            breakpoints = response["body"]["breakpoints"]
+        session = self.build_and_create_session()
+        with session.configure(LaunchArgs(self.getBuildArtifact("a.out"))):
+            functions = ["twelve"]
+            # Set a function breakpoint at 'twelve'.
+            response = session.set_function_breakpoints(functions)
+            breakpoints = response.body.breakpoints
             self.assertEqual(
                 len(breakpoints),
                 len(functions),
-                "expect %u source breakpoints" % (len(functions)),
+                f"expect {len(functions)} source breakpoints",
             )
-            for breakpoint in breakpoints:
-                bp_id_12 = breakpoint["id"]
-                self.assertTrue(breakpoint["verified"], "expect breakpoint 
verified")
-
-        # Add an extra name and make sure we have two breakpoints after this
-        functions.append("thirteen")
-        response = self.dap_server.request_setFunctionBreakpoints(functions)
-        if response:
-            breakpoints = response["body"]["breakpoints"]
+            bp_id_12 = self.expect_not_none(breakpoints[0].id)
+            self.assertTrue(breakpoints[0].verified, "expect breakpoint 
verified")
+
+            # Add an extra name and make sure we have two breakpoints after 
this.
+            functions.append("thirteen")
+            response = session.set_function_breakpoints(functions)
+            breakpoints = response.body.breakpoints
             self.assertEqual(
                 len(breakpoints),
                 len(functions),
-                "expect %u source breakpoints" % (len(functions)),
+                f"expect {len(functions)} source breakpoints",
             )
-            for breakpoint in breakpoints:
-                self.assertTrue(breakpoint["verified"], "expect breakpoint 
verified")
-
-        # There is no breakpoint delete packet, clients just send another
-        # setFunctionBreakpoints packet with the different function names.
-        functions.remove("thirteen")
-        response = self.dap_server.request_setFunctionBreakpoints(functions)
-        if response:
-            breakpoints = response["body"]["breakpoints"]
+            for bp in breakpoints:
+                self.assertTrue(bp.verified, "expect breakpoint verified")
+
+            # There is no breakpoint delete packet, clients just send another
+            # setFunctionBreakpoints packet with the different function names.
+            functions.remove("thirteen")
+            response = session.set_function_breakpoints(functions)
+            breakpoints = response.body.breakpoints
             self.assertEqual(
                 len(breakpoints),
                 len(functions),
-                "expect %u source breakpoints" % (len(functions)),
+                f"expect {len(functions)} source breakpoints",
             )
-            for breakpoint in breakpoints:
-                bp_id = breakpoint["id"]
+            for bp in breakpoints:
                 self.assertEqual(
-                    bp_id, bp_id_12, 'verify "twelve" breakpoint ID is same'
-                )
-                self.assertTrue(
-                    breakpoint["verified"], "expect breakpoint still verified"
+                    bp.id, bp_id_12, 'verify "twelve" breakpoint ID is same'
                 )
-
-        # Now get the full list of breakpoints set in the target and verify
-        # we have only 1 breakpoints set. The response above could have told
-        # us about 1 breakpoints, but we want to make sure we don't have the
-        # second one still set in the target
-        response = self.dap_server.request_testGetTargetBreakpoints()
-        if response:
-            breakpoints = response["body"]["breakpoints"]
+                self.assertTrue(bp.verified, "expect breakpoint still 
verified")
+
+            # Now get the full list of breakpoints set in the target and verify
+            # we have only 1 breakpoints set. The response above could have 
told
+            # us about 1 breakpoints, but we want to make sure we don't have 
the
+            # second one still set in the target.
+            response = 
session.send_request(DAPTestGetTargetBreakpointsArgs()).result()
+            breakpoints = response.body.breakpoints
             self.assertEqual(
                 len(breakpoints),
                 len(functions),
-                "expect %u source breakpoints" % (len(functions)),
+                f"expect {len(functions)} source breakpoints",
             )
-            for breakpoint in breakpoints:
-                bp_id = breakpoint["id"]
+            for bp in breakpoints:
                 self.assertEqual(
-                    bp_id, bp_id_12, 'verify "twelve" breakpoint ID is same'
-                )
-                self.assertTrue(
-                    breakpoint["verified"], "expect breakpoint still verified"
+                    bp.id, bp_id_12, 'verify "twelve" breakpoint ID is same'
                 )
+                self.assertTrue(bp.verified, "expect breakpoint still 
verified")
 
-        # Now clear all breakpoints for the source file by passing down an
-        # empty lines array
-        functions = []
-        response = self.dap_server.request_setFunctionBreakpoints(functions)
-        if response:
-            breakpoints = response["body"]["breakpoints"]
+            # Now clear all breakpoints for the source file by passing down an
+            # empty lines array.
+            functions = []
+            response = session.set_function_breakpoints(functions)
+            breakpoints = response.body.breakpoints
             self.assertEqual(
                 len(breakpoints),
                 len(functions),
-                "expect %u source breakpoints" % (len(functions)),
+                f"expect {len(functions)} source breakpoints",
             )
 
-        # Verify with the target that all breakpoints have been cleared
-        response = self.dap_server.request_testGetTargetBreakpoints()
-        if response:
-            breakpoints = response["body"]["breakpoints"]
+            # Verify with the target that all breakpoints have been cleared.
+            response = 
session.send_request(DAPTestGetTargetBreakpointsArgs()).result()
+            breakpoints = response.body.breakpoints
             self.assertEqual(
                 len(breakpoints),
                 len(functions),
-                "expect %u source breakpoints" % (len(functions)),
+                f"expect {len(functions)} source breakpoints",
             )
 
     @skipIfWindows
     def test_functionality(self):
         """Tests hitting breakpoints and the functionality of a single
         breakpoint, like 'conditions' and 'hitCondition' settings."""
-
+        session = self.build_and_create_session()
         program = self.getBuildArtifact("a.out")
-        self.build_and_launch(program)
-        # Set a breakpoint on "twelve" with no condition and no hitCondition
-        functions = ["twelve"]
-        breakpoint_ids = self.set_function_breakpoints(functions)
 
-        self.assertEqual(len(breakpoint_ids), len(functions), "expect one 
breakpoint")
+        # Set a breakpoint on "twelve" with no condition and no hitCondition.
+        with session.configure(LaunchArgs(program)) as ctx:
+            [bp_id] = session.resolve_function_breakpoints(["twelve"])
 
-        # Verify we hit the breakpoint we just set
-        self.continue_to_breakpoints(breakpoint_ids)
+        # Verify we hit the breakpoint we just set.
+        stop_event = session.verify_stopped_on_breakpoint(
+            bp_id, after=ctx.process_event
+        )
 
-        # Make sure i is zero at first breakpoint
-        i = int(self.dap_server.get_local_variable_value("i"))
+        # Make sure i is zero at first breakpoint.
+        thread_ctx = session.thread_context_from(stop_event)
+        i = thread_ctx.top_frame().locals["i"].value_as_int
         self.assertEqual(i, 0, "i != 0 after hitting breakpoint")
 
-        # Update the condition on our breakpoint
-        new_breakpoint_ids = self.set_function_breakpoints(functions, 
condition="i==4")
+        # Update the condition on our breakpoint.
+        func_bp = FunctionBreakpoint(name="twelve", condition="i==4")
+        [condition_bp_id] = session.resolve_function_breakpoints([func_bp])
         self.assertEqual(
-            breakpoint_ids,
-            new_breakpoint_ids,
-            "existing breakpoint should have its condition " "updated",
+            bp_id,
+            condition_bp_id,
+            "existing breakpoint should have its condition updated",
         )
 
-        self.continue_to_breakpoints(breakpoint_ids)
-        i = int(self.dap_server.get_local_variable_value("i"))
+        session.continue_to_breakpoint(bp_id)
+        i = thread_ctx.top_frame().locals["i"].value_as_int
         self.assertEqual(i, 4, "i != 4 showing conditional works")
-        new_breakpoint_ids = self.set_function_breakpoints(functions, 
hitCondition="2")
 
+        func_bp = FunctionBreakpoint(name="twelve", hitCondition="2")
+        [hit_condition_bp_id] = session.resolve_function_breakpoints([func_bp])
         self.assertEqual(
-            breakpoint_ids,
-            new_breakpoint_ids,
-            "existing breakpoint should have its condition " "updated",
+            bp_id,
+            hit_condition_bp_id,
+            "existing breakpoint should have its condition updated",
         )
 
-        # Continue with a hitCondition of 2 and expect it to skip 1 value
-        self.continue_to_breakpoints(breakpoint_ids)
-        i = int(self.dap_server.get_local_variable_value("i"))
+        # Continue with a hitCondition of 2 and expect it to skip 1 value.
+        session.continue_to_breakpoint(bp_id)
+        i = thread_ctx.top_frame().locals["i"].value_as_int
         self.assertEqual(i, 6, "i != 6 showing hitCondition works")
 
-        # continue after hitting our hitCondition and make sure it only goes
-        # up by 1
-        self.continue_to_breakpoints(breakpoint_ids)
-        i = int(self.dap_server.get_local_variable_value("i"))
+        # Continue after hitting our hitCondition and make sure it only goes
+        # up by 1.
+        session.continue_to_breakpoint(bp_id)
+        i = thread_ctx.top_frame().locals["i"].value_as_int
         self.assertEqual(i, 7, "i != 7 showing post hitCondition hits every 
time")

_______________________________________________
lldb-commits mailing list
[email protected]
https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits

Reply via email to