[Lldb-commits] [lldb] bd3546e - [lldb-dap] Migrate DAP attach tests. (#210814)
via lldb-commits
lldb-commits at lists.llvm.org
Thu Jul 23 05:52:29 PDT 2026
Author: Ebuka Ezike
Date: 2026-07-23T13:52:25+01:00
New Revision: bd3546eaedaa0bc89f6706c619074d59ba00bafe
URL: https://github.com/llvm/llvm-project/commit/bd3546eaedaa0bc89f6706c619074d59ba00bafe
DIFF: https://github.com/llvm/llvm-project/commit/bd3546eaedaa0bc89f6706c619074d59ba00bafe.diff
LOG: [lldb-dap] Migrate DAP attach tests. (#210814)
Address some issues with the previous tests.
- Always wait for the continued event after sending a continue request.
since the continue response is just an acknowlegement that we send a
continue packet.
- Retry reading stdin if it has an error when the debugger attaches.
- Update and enable the attachByPortNum test, this may now run on NetBSD
and Windows. will try to enable in a different PR.
Added:
Modified:
lldb/packages/Python/lldbsuite/test/tools/lldb_dap/session_helpers.py
lldb/packages/Python/lldbsuite/test/tools/lldb_dap/testcase.py
lldb/packages/Python/lldbsuite/test/tools/lldb_dap/types.py
lldb/test/API/tools/lldb-dap/attach-commands/TestDAP_attachCommands.py
lldb/test/API/tools/lldb-dap/attach-commands/main.c
lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py
lldb/test/API/tools/lldb-dap/attach/TestDAP_attachByPortNum.py
lldb/test/API/tools/lldb-dap/attach/main.c
Removed:
################################################################################
diff --git a/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/session_helpers.py b/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/session_helpers.py
index 613bf70d53479..32937ebdc4861 100644
--- a/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/session_helpers.py
+++ b/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/session_helpers.py
@@ -27,6 +27,7 @@
CompletionsArgs,
ConfigurationDoneArgs,
ContinueArgs,
+ ContinuedEvent,
DataBreakpoint,
DataBreakpointInfoArgs,
DisassembleArgs,
@@ -1103,8 +1104,17 @@ def wait_for_memory_event(self, *, after: Event | Response):
return self.wait_for_event(MemoryEvent, after=after)
def do_continue(self):
+ """Send a 'continueRequest' and wait for a 'ContinuedEvent',
+
+ Receiving the continue response does not mean the process continued,
+ It means we successfully sent the continue packet.
+ LLDB can continue asynchronously, so wait for the 'Continued' event.
+ """
self.ensure_initialized()
- return self.send_request(ContinueArgs()).result()
+ prior_event = self.last_event()
+ response = self.send_request(ContinueArgs()).result()
+ self.wait_for_event(ContinuedEvent, after=prior_event)
+ return response
def continue_to_exit(self, exitCode: int = 0) -> ExitedEvent:
continue_response = self.do_continue()
diff --git a/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/testcase.py b/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/testcase.py
index 18085e5156aef..9e43a100cbd56 100644
--- a/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/testcase.py
+++ b/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/testcase.py
@@ -1,12 +1,14 @@
import logging
import os
+import uuid
from pathlib import Path
from typing import Any, Final, Optional, TypeVar, Union, cast
from lldbsuite.test.lldbtest import Base, LLDBTestCaseFactory, is_exe
+import lldbgdbserverutils
-from .types import AnyResponse, ErrorResponse, Response
from .session_helpers import DAPTestSession
+from .types import AnyResponse, ErrorResponse, Response
from .utils import DebugAdapter, DebugAdapterOptions
@@ -116,6 +118,11 @@ def cleanup_session():
self.addTearDownHook(cleanup_session)
return session
+ def build_for_attach(self) -> str:
+ unique_name = str(uuid.uuid4())
+ self.build(dictionary={"EXE": unique_name})
+ return self.getBuildArtifact(unique_name)
+
def build_and_create_session(
self,
adapter: Optional[DebugAdapter] = None,
@@ -152,6 +159,7 @@ def create_debug_adapter(
def cleanup_adapter():
if adapter.is_alive:
+ self.logger.info("Manually killing debug adapter.")
adapter.kill()
# The debug adapter may have a reason the test failed.
if stderr := adapter.process.stderr:
@@ -194,6 +202,24 @@ def create_server_debug_adapter(
self.assertTrue(adapter.is_server, "adapter should run as a server.")
return adapter
+ def get_debug_server_path(self) -> Optional[Path]:
+ # Tries to find simulation/lldb-server/gdbserver tool path.
+ platform = self.getPlatform()
+ if platform == "windows":
+ # TODO: Update this for windows as it now supports lldb-server.
+ return None
+
+ if self.platformIsDarwin():
+ if platform != "macosx":
+ return None
+ server_exe = lldbgdbserverutils.get_debugserver_exe()
+ else:
+ server_exe = lldbgdbserverutils.get_lldb_server_exe()
+
+ server_path = Path(server_exe)
+ self.assertTrue(server_path.exists(), f"{server_path.stem!r} not found")
+ return server_path
+
def expect_not_none(self, value: Optional[T], msg: Any = None) -> T:
"""Convenience function to narrow fields that are optional, as most DAP types are."""
self.assertIsNotNone(value, msg=msg)
diff --git a/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/types.py b/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/types.py
index cfd14f480e2a6..19bd0d9dc26e5 100644
--- a/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/types.py
+++ b/lldb/packages/Python/lldbsuite/test/tools/lldb_dap/types.py
@@ -1150,15 +1150,24 @@ class LaunchArgs:
@dataclass(frozen=True)
@args_protocol
class AttachArgs:
- restart: Optional[Any] = field(metadata={"alias": "__restart"}, default=None)
+ @dataclass(frozen=True)
+ class Session:
+ targetId: int
+ debuggerId: Optional[int] = None
+ program: Optional[str] = None
attachCommands: Optional[List[str]] = None
pid: Optional[int] = None
waitFor: Optional[bool] = None
- gdbRemotePort: Optional[int] = None
- gdbRemoteHostname: Optional[str] = None
+ gdbRemotePort: Optional[int] = field(
+ metadata={"alias": "gdb-remote-port"}, default=None
+ )
+ gdbRemoteHostname: Optional[str] = field(
+ metadata={"alias": "gdb-remote-hostname"}, default=None
+ )
coreFile: Optional[str] = None
- program: Optional[str] = None
+ session: Optional[Session] = None
+ restart: Optional[Any] = field(metadata={"alias": "__restart"}, default=None)
# Configurations.
debuggerRoot: Optional[str] = None
diff --git a/lldb/test/API/tools/lldb-dap/attach-commands/TestDAP_attachCommands.py b/lldb/test/API/tools/lldb-dap/attach-commands/TestDAP_attachCommands.py
index f24a32edbaa7b..0cfbb30f02c31 100644
--- a/lldb/test/API/tools/lldb-dap/attach-commands/TestDAP_attachCommands.py
+++ b/lldb/test/API/tools/lldb-dap/attach-commands/TestDAP_attachCommands.py
@@ -2,16 +2,12 @@
Test lldb-dap attach commands
"""
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
-import time
+from lldbsuite.test.decorators import skipIfNetBSD
+from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase
+from lldbsuite.test.tools.lldb_dap.types import AttachArgs, PauseArgs
-class TestDAP_attachCommands(lldbdap_testcase.DAPTestCaseBase):
- SHARED_BUILD_TESTCASE = False
-
+class TestDAP_attachCommands(DAPTestCaseBase):
@skipIfNetBSD # Hangs on NetBSD as well
def test_commands(self):
"""
@@ -34,89 +30,97 @@ def test_commands(self):
"terminateCommands" are a list of LLDB commands that get executed when
the debugger session terminates.
"""
- program = self.build_and_create_debug_adapter_for_attach()
+ program = self.getBuildArtifact("a.out")
+ session = self.build_and_create_session()
# Here we just create a target and launch the process as a way to test
# if we are able to use attach commands to create any kind of a target
- # and use it for debugging
+ # and use it for debugging.
attachCommands = [
- 'target create -d "%s"' % (program),
- "process launch --stop-at-entry",
+ f'target create -d "{program}"',
+ "process launch --stop-at-user-entry",
]
initCommands = ["target list", "platform list"]
preRunCommands = ["image list a.out", "image dump sections a.out"]
postRunCommands = ["help trace", "help process trace"]
stopCommands = ["frame variable", "thread backtrace"]
- exitCommands = ["expr 2+3", "expr 3+4"]
- terminateCommands = ["expr 4+2"]
- self.attach(
- program=program,
- attachCommands=attachCommands,
- initCommands=initCommands,
- preRunCommands=preRunCommands,
- stopCommands=stopCommands,
- exitCommands=exitCommands,
- terminateCommands=terminateCommands,
- postRunCommands=postRunCommands,
+ exitCommands = ["history -c 2"]
+ terminateCommands = ["platform status"]
+
+ process_event = session.attach(
+ AttachArgs(
+ program=program,
+ attachCommands=attachCommands,
+ stopOnEntry=True,
+ initCommands=initCommands,
+ preRunCommands=preRunCommands,
+ stopCommands=stopCommands,
+ exitCommands=exitCommands,
+ terminateCommands=terminateCommands,
+ postRunCommands=postRunCommands,
+ )
)
- self.dap_server.wait_for_initialized()
+
# Get output from the console. This should contain both the
# "initCommands" and the "preRunCommands".
- output = self.get_console()
+ output = session.get_console()
# Verify all "initCommands" were found in console output
- self.verify_commands("initCommands", output, initCommands)
+ session.verify_commands("initCommands", output, initCommands)
# Verify all "preRunCommands" were found in console output
- self.verify_commands("preRunCommands", output, preRunCommands)
+ session.verify_commands("preRunCommands", output, preRunCommands)
# Verify all "postRunCommands" were found in console output
- self.verify_commands("postRunCommands", output, postRunCommands)
+ session.verify_commands("postRunCommands", output, postRunCommands)
- functions = ["main"]
- breakpoint_ids = self.set_function_breakpoints(functions)
- self.assertEqual(len(breakpoint_ids), len(functions), "expect one breakpoint")
- self.continue_to_breakpoints(breakpoint_ids)
- output = self.collect_console(pattern=stopCommands[-1])
- self.verify_commands("stopCommands", output, stopCommands)
+ stopped_event = session.verify_stopped_on_entry(after=process_event)
+ stopped_thread_id = self.expect_not_none(stopped_event.body.threadId)
- # Check that we got module events from target
- modules = self.dap_server.wait_for_module_events()
- self.assertGreater(len(modules), 0)
+ output = session.collect_console(after=stopped_event, until=stopCommands[-1])
+ session.verify_commands("stopCommands", output.seen_texts, stopCommands)
# Continue after launch and hit the "pause()" call and stop the target.
# Get output from the console. This should contain both the
# "stopCommands" that were run after we stop.
- self.do_continue()
- time.sleep(0.5)
- self.dap_server.request_pause()
- self.dap_server.wait_for_stopped()
- output = self.collect_console(pattern=stopCommands[-1])
- self.verify_commands("stopCommands", output, stopCommands)
-
- # Continue until the program exits
- self.continue_to_exit()
+ session.do_continue()
+
+ before_pause = session.last_event()
+ session.send_request(PauseArgs(stopped_thread_id)).result()
+ session.wait_for_stopped_event(after=before_pause)
+
+ output = session.collect_console(after=before_pause, until=stopCommands[-1])
+ session.verify_commands("stopCommands", output.seen_texts, stopCommands)
+
+ # Set is_ready and continue until the program exits.
+ session.evaluate("`expression is_ready = 1", context="repl")
+ session.continue_to_exit()
+
# Get output from the console. This should contain both the
# "exitCommands" that were run after the second breakpoint was hit
# and the "terminateCommands" due to the debugging session ending
- output = self.collect_console(
- pattern=terminateCommands[0],
- )
- self.verify_commands("exitCommands", output, exitCommands)
- self.verify_commands("terminateCommands", output, terminateCommands)
+ output = session.collect_console(after=output.event, until=terminateCommands[0])
+ output_texts = output.seen_texts
+ session.verify_commands("exitCommands", output_texts, exitCommands)
+ session.verify_commands("terminateCommands", output_texts, terminateCommands)
def test_attach_command_process_failures(self):
"""
Tests that a 'attachCommands' is expected to leave the debugger's
selected target with a valid process.
"""
- program = self.build_and_create_debug_adapter_for_attach()
- attachCommands = ['script print("oops, forgot to attach to a process...")']
- resp = self.attach_and_configurationDone(
+ program = self.getBuildArtifact("a.out")
+ session = self.build_and_create_session()
+
+ attach_args = AttachArgs(
program=program,
- attachCommands=attachCommands,
+ attachCommands=['script print("oops, forgot to attach to a process...")'],
)
- self.assertFalse(resp["success"])
+ pending_attach = session.send_request(attach_args)
+ session.verify_configuration_done(expected_success=False)
+
+ attach_response = pending_attach.error()
+ response_body = self.expect_not_none(attach_response.body)
+ response_error = self.expect_not_none(response_body.error)
self.assertIn(
- "attachCommands failed to attach to a process",
- resp["body"]["error"]["format"],
+ "attachCommands failed to attach to a process", response_error.format
)
@skipIfNetBSD # Hangs on NetBSD as well
@@ -125,27 +129,29 @@ def test_terminate_commands(self):
Tests that the "terminateCommands", that can be passed during
attach, are run when the debugger is disconnected.
"""
- program = self.build_and_create_debug_adapter_for_attach()
+ program = self.getBuildArtifact("a.out")
+ session = self.build_and_create_session(disconnect_automatically=False)
# Here we just create a target and launch the process as a way to test
# if we are able to use attach commands to create any kind of a target
# and use it for debugging
attachCommands = [
- 'target create -d "%s"' % (program),
- "process launch --stop-at-entry",
+ f"target create -d '{program}'",
+ "process launch --stop-at-user-entry",
]
- terminateCommands = ["expr 4+2"]
- self.attach(
- program=program,
- attachCommands=attachCommands,
- terminateCommands=terminateCommands,
- disconnectAutomatically=False,
+ terminateCommands = ["history -c 1"]
+ process_event = session.attach(
+ AttachArgs(
+ program=program,
+ attachCommands=attachCommands,
+ terminateCommands=terminateCommands,
+ )
+ )
+ # Once it's disconnected the console should contain the "terminateCommands".
+ session.disconnect(terminateDebuggee=True)
+ output = session.collect_console(
+ after=process_event, until=terminateCommands[0]
)
- self.get_console()
- # Once it's disconnected the console should contain the
- # "terminateCommands"
- self.dap_server.request_disconnect(terminateDebuggee=True)
- output = self.collect_console(
- pattern=terminateCommands[0],
+ session.verify_commands(
+ "terminateCommands", output.seen_texts, terminateCommands
)
- self.verify_commands("terminateCommands", output, terminateCommands)
diff --git a/lldb/test/API/tools/lldb-dap/attach-commands/main.c b/lldb/test/API/tools/lldb-dap/attach-commands/main.c
index f56d5d53afa05..26bb99c93cc84 100644
--- a/lldb/test/API/tools/lldb-dap/attach-commands/main.c
+++ b/lldb/test/API/tools/lldb-dap/attach-commands/main.c
@@ -1,30 +1,21 @@
#include "attach.h"
-#include <stdio.h>
+
#ifdef _WIN32
-#include <process.h>
#include <windows.h>
+#define sleep_ms(ms) Sleep(ms)
#else
#include <unistd.h>
+#define sleep_ms(ms) usleep((ms) * 1000)
#endif
+static volatile int is_ready = 0;
+
int main(int argc, char const *argv[]) {
lldb_enable_attach();
- if (argc >= 2) {
- // Create the synchronization token.
- FILE *f = fopen(argv[1], "wx");
- if (!f)
- return 1;
- fputs("\n", f);
- fflush(f);
- fclose(f);
+ while (!is_ready) {
+ sleep_ms(50);
}
- printf("pid = %i\n", getpid());
-#ifdef _WIN32
- Sleep(10 * 1000);
-#else
- sleep(10);
-#endif
- return 0; // breakpoint 1
+ return 0;
}
diff --git a/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py b/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py
index 8e8fea5f65b4b..96c5b193df39f 100644
--- a/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py
+++ b/lldb/test/API/tools/lldb-dap/attach/TestDAP_attach.py
@@ -2,77 +2,79 @@
Test lldb-dap attach request
"""
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbutil
-import lldbdap_testcase
-import os
import subprocess
-import threading
-import time
+import uuid
+from pathlib import Path
+
+from lldbsuite.test import lldbutil
+from lldbsuite.test.decorators import (
+ expectedFailureWindows,
+ expectedFailureWindowsAndNoLLDBServer,
+ skipIf,
+ skipIfWasm,
+ skipIfWindowsAndLLDBServer,
+)
+from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase
+from lldbsuite.test.tools.lldb_dap.types import (
+ AttachArgs,
+ ProcessEvent,
+ ProgressStartEvent,
+)
# Often fails on Arm Linux, but not specifically because it's Arm, something in
# process scheduling can cause a massive (minutes) delay during this test.
@skipIf(oslist=["linux"], archs=["arm$"])
- at skipIfWasm # no attach support
-class TestDAP_attach(lldbdap_testcase.DAPTestCaseBase):
+ at skipIfWasm # No attach support
+class TestDAP_attach(DAPTestCaseBase):
SHARED_BUILD_TESTCASE = False
- def spawn(self, program, args=None):
- return self.spawnSubprocess(
+ def spawn(self, program: str, *, wait_for_sync: bool = True):
+ """Spawn the target and (by default) block until it has called `lldb_enable_attach`."""
+ sync_token = lldbutil.append_to_process_working_directory(
+ self, f"sync_{uuid.uuid4().hex}"
+ )
+ proc = self.spawnSubprocess(
executable=program,
- args=args,
+ args=[sync_token],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
)
+ if wait_for_sync:
+ lldbutil.wait_for_file_on_target(self, sync_token)
- def spawn_and_wait(self, program, delay):
- time.sleep(delay)
- proc = self.spawn(program=program)
- start_time = time.time()
- # Wait for either the process to exit or the event to be set.
- while proc.poll() is None and not self.spawn_event.is_set():
- elapsed = time.time() - start_time
- if elapsed >= self.DEFAULT_TIMEOUT:
- break
- time.sleep(0.1)
- proc.kill()
- proc.wait()
-
- def continue_and_verify_pid(self):
- self.do_continue()
- proc = self.lastSubprocess
- if proc is None:
- self.fail(f"lastSubprocess is None")
- out, _ = proc.communicate("foo")
+ self.subprocesses.append(proc)
+ return proc
+
+ def verify_pid(self, proc):
+ out, _ = proc.communicate("f")
+
+ self.assertIn(f"char = f", out)
self.assertIn(f"pid = {proc.pid}", out)
def test_by_pid(self):
- """
- Tests attaching to a process by process ID.
- """
- program = self.build_and_create_debug_adapter_for_attach()
+ """Tests attaching to a process by process ID."""
+ program = self.build_for_attach()
+ session = self.create_session()
+
proc = self.spawn(program=program)
- self.attach(pid=proc.pid)
- self.continue_and_verify_pid()
+ self.assertIsNone(proc.poll(), "process should be running")
+
+ process_event = session.attach(AttachArgs(pid=proc.pid))
+ self.assertEqual(process_event.body.systemProcessId, proc.pid)
+ self.verify_pid(proc)
def test_by_name(self):
- """
- Tests attaching to a process by process name.
- """
- program = self.build_and_create_debug_adapter_for_attach()
+ """Tests attaching to a process by process name."""
+ program = self.build_for_attach()
+ session = self.create_session()
- # Use a file as a synchronization point between test and inferior.
- pid_file_path = lldbutil.append_to_process_working_directory(
- self, "pid_file_%d" % (int(time.time()))
- )
- self.spawn(program=program, args=[pid_file_path])
- lldbutil.wait_for_file_on_target(self, pid_file_path)
+ proc = self.spawn(program=program)
- self.attach(program=program)
- self.continue_and_verify_pid()
+ process_event = session.attach(AttachArgs(program=program))
+ self.assertEqual(process_event.body.systemProcessId, proc.pid)
+ self.verify_pid(proc)
@expectedFailureWindowsAndNoLLDBServer
def test_by_name_waitFor(self):
@@ -80,64 +82,63 @@ def test_by_name_waitFor(self):
Tests waiting for, and attaching to a process by process name that
doesn't exist yet.
"""
- program = self.build_and_create_debug_adapter_for_attach()
- self.spawn_event = threading.Event()
- self.spawn_thread = threading.Thread(
- target=self.spawn_and_wait,
- args=(
- program,
- 1.0,
- ),
- )
- self.spawn_thread.start()
- try:
- self.attach(program=program, waitFor=True)
- self.continue_and_verify_pid()
- finally:
- self.spawn_event.set()
- if self.spawn_thread.is_alive():
- self.spawn_thread.join(timeout=10)
+ self.do_attach_waitFor(use_basename=False)
@expectedFailureWindows
@skipIfWindowsAndLLDBServer
- def test_by_partial_name_waitFor(self):
+ def test_by_basename_waitFor(self):
"""
- Tests waiting for and attaching to a process by partial process name
+ Tests waiting for and attaching to a process by the process base name
that doesn't exist yet.
"""
- program = self.build_and_create_debug_adapter_for_attach()
- self.spawn_event = threading.Event()
- self.spawn_thread = threading.Thread(
- target=self.spawn_and_wait,
- args=(
- program,
- 1.0,
- ),
+ self.do_attach_waitFor(use_basename=True)
+
+ def do_attach_waitFor(self, use_basename: bool):
+ """Kick off attach with waitFor=True; spawn the target once lldb-dap
+ signals it has entered the wait-for-process polling loop."""
+ session = self.create_session()
+ program = self.build_for_attach()
+ attach_name = Path(program).name if use_basename else program
+
+ init_response = session.initialize_sequence(session.initialize_args)
+ pending = session.send_request(AttachArgs(program=attach_name, waitFor=True))
+
+ # Wait until lldb-dap is actually polling for the process before we
+ # spawn it, so we don't race the poll setup.
+ session.wait_for_event(
+ ProgressStartEvent,
+ until=lambda e: "Waiting to attach" in e.body.title,
+ after=init_response,
+ timeout_msg="Waiting for attach progress event.",
)
- self.spawn_thread.start()
- try:
- self.attach(program=os.path.basename(program), waitFor=True)
- self.continue_and_verify_pid()
- finally:
- self.spawn_event.set()
- if self.spawn_thread.is_alive():
- self.spawn_thread.join(timeout=10)
+
+ proc = self.spawn(program=program, wait_for_sync=False)
+
+ session.ensure_initialized()
+ session.verify_configuration_done()
+
+ process_event = session.wait_for_event(ProcessEvent, after=init_response)
+ pending.result("expects attach response.")
+
+ self.assertEqual(process_event.body.systemProcessId, proc.pid)
+ self.verify_pid(proc)
def test_attach_with_missing_session_debugger(self):
"""
Test that attaching with only one of debuggerId/targetId specified
fails with the expected error message.
"""
- self.build_and_create_debug_adapter()
+ session = self.create_session()
# Test with only targetId specified (no debuggerId)
- session = {"targetId": 99999}
- attach_seq = self.attach(session=session)
- resp = self.dap_server.receive_response(attach_seq)
- self.assertFalse(resp["success"])
+ resp = session.initialize_and_launch(
+ AttachArgs(session=AttachArgs.Session(targetId=99999))
+ ).error()
+
+ message = self.expect_not_none(resp.body and resp.body.error)
self.assertIn(
"missing value at arguments.session.debuggerId",
- resp["body"]["error"]["format"],
+ message.format,
)
def test_attach_with_invalid_session(self):
@@ -145,15 +146,19 @@ def test_attach_with_invalid_session(self):
Test that attaching with both debuggerId and targetId specified but
invalid fails with an appropriate error message.
"""
- self.build_and_create_debug_adapter()
+ session = self.create_session()
- # Attach with both debuggerId=9999 and targetId=99999 (both invalid).
+ # Attach with both debuggerId=9999 and targetId=9999 (both invalid).
# Since debugger ID 9999 likely doesn't exist in the global registry,
# we expect a validation error.
- session = {"debuggerId": 9999, "targetId": 9999}
- resp = self.attach_and_configurationDone(session=session)
- self.assertFalse(resp["success"])
- error_msg = resp["body"]["error"]["format"]
+ pending = session.initialize_and_launch(
+ AttachArgs(session=AttachArgs.Session(targetId=9999, debuggerId=9999))
+ )
+ session.configuration_done().result_or_error()
+
+ resp = pending.error()
+ message = self.expect_not_none(resp.body and resp.body.error)
+ error_msg = message.format
# Either error is acceptable - both indicate the debugger reuse
# validation is working correctly
self.assertTrue(
diff --git a/lldb/test/API/tools/lldb-dap/attach/TestDAP_attachByPortNum.py b/lldb/test/API/tools/lldb-dap/attach/TestDAP_attachByPortNum.py
index e8f3f0fbc3fd3..fe4f240c404e2 100644
--- a/lldb/test/API/tools/lldb-dap/attach/TestDAP_attachByPortNum.py
+++ b/lldb/test/API/tools/lldb-dap/attach/TestDAP_attachByPortNum.py
@@ -2,151 +2,118 @@
Test lldb-dap "port" configuration to "attach" request
"""
-from lldbsuite.test.decorators import *
-from lldbsuite.test.lldbtest import *
-from lldbsuite.test import lldbplatformutil
+from typing import List
+
from lldbgdbserverutils import Pipe
-import lldbdap_testcase
+from lldbsuite.test import lldbplatformutil
+from lldbsuite.test.decorators import skipIfNetBSD, skipIfWindows
+from lldbsuite.test.lldbtest import line_number
+from lldbsuite.test.tools.lldb_dap import DAPTestCaseBase
+from lldbsuite.test.tools.lldb_dap.types import AttachArgs
+
import lldb
- at skip(bugnumber="https://github.com/llvm/llvm-project/issues/138803")
-class TestDAP_attachByPortNum(lldbdap_testcase.DAPTestCaseBase):
- def set_and_hit_breakpoint(self, continueToExit=True):
- self.dap_server.wait_for_stopped()
+def debug_server_start_args() -> List[str]:
+ args: List[str] = []
+ if not lldbplatformutil.platformIsDarwin():
+ args = ["gdbserver"]
- source = "main.c"
- breakpoint1_line = line_number(source, "// breakpoint 1")
- lines = [breakpoint1_line]
- # Set breakpoint in the thread function so we can step the threads
- 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)
- if continueToExit:
- self.continue_to_exit()
-
- def get_debug_server_command_line_args(self):
- args = []
- if lldbplatformutil.getPlatform() == "linux":
- args = ["gdbserver"]
- if lldb.remote_platform:
- args += ["*:0"]
- else:
- args += ["localhost:0"]
- return args
-
- def get_debug_server_pipe(self):
+ if lldb.remote_platform:
+ args += ["*:0"]
+ else:
+ args += ["localhost:0"]
+ return args
+
+
+class TestDAP_attachByPortNum(DAPTestCaseBase):
+ SHARED_BUILD_TESTCASE = False
+
+ def create_debug_server_pipe(self):
pipe = Pipe(self.getBuildDir())
- self.addTearDownHook(lambda: pipe.close())
+ self.addTearDownHook(pipe.close)
pipe.finish_connection(self.DEFAULT_TIMEOUT)
return pipe
@skipIfWindows
- @skipIfNetBSD
+ @skipIfNetBSD # Try enable, get_debug_server_path previously returned None.
def test_by_port(self):
- """
- Tests attaching to a process by port.
- """
- program = self.build_and_create_debug_adapter_for_attach()
-
- debug_server_tool = self.getBuiltinDebugServerTool()
-
- pipe = self.get_debug_server_pipe()
- args = self.get_debug_server_command_line_args()
- args += [program]
- args += ["--named-pipe", pipe.name]
-
- self.process = self.spawnSubprocess(
- debug_server_tool, args, install_remote=False
+ """Tests attaching to a process by port."""
+ program_path = self.build_for_attach()
+ session = self.create_session()
+
+ pipe = self.create_debug_server_pipe()
+ debug_server_args = debug_server_start_args()
+ debug_server_args.extend(["--named-pipe", pipe.name, "--", program_path])
+
+ self.spawnSubprocess(
+ str(self.get_debug_server_path()),
+ debug_server_args,
+ install_remote=False,
)
# Read the port number from the debug server pipe.
- port = pipe.read(10, self.DEFAULT_TIMEOUT)
- # Trim null byte, convert to int
- port = int(port[:-1])
- self.assertIsNotNone(
- port, " Failed to read the port number from debug server pipe"
- )
+ pipe_data = pipe.read(10, self.DEFAULT_TIMEOUT)
+ port = int(pipe_data.rstrip(b"\0"))
- self.attach(
- program=program,
- gdbRemotePort=port,
- sourceInitFile=True,
- stopOnEntry=True,
- )
- self.set_and_hit_breakpoint(continueToExit=True)
+ args = AttachArgs(program=program_path, gdbRemotePort=port, stopOnEntry=True)
+ with session.configure(args) as ctx:
+ bp_line = line_number("main.c", "// breakpoint 1")
+ [bp1] = session.resolve_source_breakpoints("main.c", [bp_line])
+
+ session.verify_stopped_on_entry(after=ctx.process_event)
+ session.continue_to_breakpoint(bp1)
+ session.continue_to_exit()
@skipIfWindows
@skipIfNetBSD
def test_fails_if_both_port_and_pid_are_set(self):
- """
- Tests attaching to a process by process ID and port number.
- """
- program = self.build_and_create_debug_adapter_for_attach()
-
+ """Tests attaching to a process by process ID and port number."""
# It is not necessary to launch "lldb-server" to obtain the actual port
# and pid for attaching. However, when providing the port number and pid
# directly, "lldb-dap" throws an error message, which is expected. So,
# used random pid and port numbers here.
+ program = self.build_for_attach()
+ session = self.create_session()
- pid = 1354
- port = 1234
-
- response = self.attach(
- program=program,
- pid=pid,
- gdbRemotePort=port,
- sourceInitFile=True,
- waitForResponse=True,
- )
- self.assertFalse(
- response["success"], "The user can't specify both pid and port"
+ pending = session.send_request(
+ AttachArgs(program, pid=1354, gdbRemotePort=1234)
)
+ pending.error("The user can't specify both pid and port")
@skipIfWindows
@skipIfNetBSD
def test_by_invalid_port(self):
- """
- Tests attaching to a process by invalid port number 0.
- """
- program = self.build_and_create_debug_adapter_for_attach()
-
- port = 0
- response = self.attach(
- program=program,
- gdbRemotePort=port,
- sourceInitFile=True,
- waitForResponse=True,
- )
- self.assertFalse(
- response["success"],
- "The user can't attach with invalid port (%s)" % port,
- )
+ """Tests attaching to a process by invalid port number 0."""
+ program = self.build_for_attach()
+ session = self.create_session()
+
+ port = -1
+ attach_args = AttachArgs(program, gdbRemotePort=port)
+ pending = session.initialize_and_launch(attach_args)
+ session.configuration_done().result_or_error()
+ pending.error(f"The user can't attach to invalid port {port}")
@skipIfWindows
@skipIfNetBSD
def test_by_illegal_port(self):
- """
- Tests attaching to a process by illegal/greater port number 65536
- """
- program = self.build_and_create_debug_adapter_for_attach()
+ """Tests attaching to a process by illegal/greater port number 65536"""
+ program = self.build_for_attach()
+ session = self.create_session()
port = 65536
- args = [program]
- debug_server_tool = self.getBuiltinDebugServerTool()
- self.process = self.spawnSubprocess(
- debug_server_tool, args, install_remote=False
- )
-
- response = self.attach(
- program=program,
- gdbRemotePort=port,
- sourceInitFile=True,
- waitForResponse=True,
- )
- self.assertFalse(
- response["success"],
- "The user can't attach with illegal port (%s)" % port,
+ debug_server = self.expect_not_none(self.get_debug_server_path())
+ server_args = [f"localhost:{port}", "--", program]
+ if debug_server.stem == "lldb-server":
+ server_args = ["gdbserver", *server_args]
+
+ self.spawnSubprocess(str(debug_server), server_args, install_remote=False)
+
+ pending = session.initialize_and_launch(
+ AttachArgs(
+ program=program,
+ gdbRemotePort=port,
+ )
)
+ session.configuration_done().result_or_error()
+ pending.error(f"The user can't attach with illegal port ({port})")
diff --git a/lldb/test/API/tools/lldb-dap/attach/main.c b/lldb/test/API/tools/lldb-dap/attach/main.c
index e14cf71a7044c..23e358de55747 100644
--- a/lldb/test/API/tools/lldb-dap/attach/main.c
+++ b/lldb/test/API/tools/lldb-dap/attach/main.c
@@ -17,11 +17,22 @@ int main(int argc, char const *argv[]) {
fputs("\n", f);
fflush(f);
fclose(f);
- }
- // Wait on input from stdin.
- getchar();
+ // Wait on input from stdin.
+ // when lldb connects to the process, on MacOS getchar() is interupted
+ // and sets the stream's error indicator (EINTR).
+ // ignore that and keep waiting until we actually receive a character.
+ while (1) {
+ int c = getchar();
+ if (c == EOF && ferror(stdin)) {
+ clearerr(stdin);
+ continue;
+ }
+ printf("char = %c\n", c);
+ break;
+ }
+ }
printf("pid = %i\n", getpid());
- return 0;
+ return 0; // breakpoint 1
}
More information about the lldb-commits
mailing list