clang-p2996/lldb/test/API/tools/lldb-dap/attach/TestDAP_attachByPortNum.py

"""
Test lldb-dap "port" configuration to "attach" request
"""


import dap_server
from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
from lldbsuite.test import lldbutil
from lldbsuite.test import lldbplatformutil
from lldbgdbserverutils import Pipe
import lldbdap_testcase
import os
import shutil
import subprocess
import tempfile
import threading
import sys
import socket


class TestDAP_attachByPortNum(lldbdap_testcase.DAPTestCaseBase):
    default_timeout = 20

    def set_and_hit_breakpoint(self, continueToExit=True):
        source = "main.c"
        main_source_path = os.path.join(os.getcwd(), source)
        breakpoint1_line = line_number(main_source_path, "// breakpoint 1")
        lines = [breakpoint1_line]
        # Set breakpoint in the thread function so we can step the threads
        breakpoint_ids = self.set_source_breakpoints(main_source_path, 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"]
        elif lldbplatformutil.getPlatform() == "macosx":
            args = ["--listen"]
        if lldb.remote_platform:
            args += ["*:0"]
        else:
            args += ["localhost:0"]
        return args

    def get_debug_server_pipe(self):
        pipe = Pipe(self.getBuildDir())
        self.addTearDownHook(lambda: pipe.close())
        pipe.finish_connection(self.default_timeout)
        return pipe

    @skipIfWindows
    @skipIfNetBSD
    def test_by_port(self):
        """
        Tests attaching to a process by port.
        """
        self.build_and_create_debug_adapter()
        program = self.getBuildArtifact("a.out")

        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
        )

        # 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"
        )

        self.attach(program=program, gdbRemotePort=port, sourceInitFile=True)
        self.set_and_hit_breakpoint(continueToExit=True)
        self.process.terminate()

    @skipIfWindows
    @skipIfNetBSD
    def test_by_port_and_pid(self):
        """
        Tests attaching to a process by process ID and port number.
        """
        self.build_and_create_debug_adapter()
        program = self.getBuildArtifact("a.out")

        # 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.

        pid = 1354
        port = 1234

        response = self.attach(
            program=program,
            pid=pid,
            gdbRemotePort=port,
            sourceInitFile=True,
            expectFailure=True,
        )
        if not (response and response["success"]):
            self.assertFalse(
                response["success"], "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.
        """
        self.build_and_create_debug_adapter()
        program = self.getBuildArtifact("a.out")

        port = 0
        response = self.attach(
            program=program, gdbRemotePort=port, sourceInitFile=True, expectFailure=True
        )
        if not (response and response["success"]):
            self.assertFalse(
                response["success"],
                "The user can't attach with invalid port (%s)" % port,
            )

    @skipIfWindows
    @skipIfNetBSD
    def test_by_illegal_port(self):
        """
        Tests attaching to a process by illegal/greater port number 65536
        """
        self.build_and_create_debug_adapter()
        program = self.getBuildArtifact("a.out")

        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, expectFailure=True
        )
        if not (response and response["success"]):
            self.assertFalse(
                response["success"],
                "The user can't attach with illegal port (%s)" % port,
            )
        self.process.terminate()