1from __future__ import print_function
2
3import gdbremote_testcase
4import select
5import socket
6from lldbsuite.test.decorators import *
7from lldbsuite.test.lldbtest import *
8
9
10class TestGdbRemoteConnection(gdbremote_testcase.GdbRemoteTestCaseBase):
11
12    mydir = TestBase.compute_mydir(__file__)
13
14    @skipIfRemote  # reverse connect is not a supported use case for now
15    def test_reverse_connect_llgs(self):
16        # Reverse connect is the default connection method.
17        self.connect_to_debug_monitor()
18        # Verify we can do the handshake.  If that works, we'll call it good.
19        self.do_handshake(self.sock)
20
21    @skipIfRemote
22    @skipIfWindows
23    def test_named_pipe_llgs(self):
24        family, type, proto, _, addr = socket.getaddrinfo(
25            self.stub_hostname, 0, proto=socket.IPPROTO_TCP)[0]
26        self.sock = socket.socket(family, type, proto)
27        self.sock.settimeout(self.DEFAULT_TIMEOUT)
28
29        self.addTearDownHook(lambda: self.sock.close())
30
31        named_pipe_path = self.getBuildArtifact("stub_port_number")
32
33        # Create the named pipe.
34        os.mkfifo(named_pipe_path)
35
36        # Open the read side of the pipe in non-blocking mode.  This will
37        # return right away, ready or not.
38        named_pipe_fd = os.open(named_pipe_path, os.O_RDONLY | os.O_NONBLOCK)
39
40        self.addTearDownHook(lambda: os.close(named_pipe_fd))
41
42        args = self.debug_monitor_extra_args
43        if lldb.remote_platform:
44            args += ["*:0"]
45        else:
46            args += ["localhost:0"]
47
48        args += ["--named-pipe", named_pipe_path]
49
50        server = self.spawnSubprocess(
51            self.debug_monitor_exe,
52            args,
53            install_remote=False)
54
55        (ready_readers, _, _) = select.select(
56            [named_pipe_fd], [], [], self.DEFAULT_TIMEOUT)
57        self.assertIsNotNone(
58            ready_readers,
59            "write side of pipe has not written anything - stub isn't writing to pipe.")
60        port = os.read(named_pipe_fd, 10)
61        # Trim null byte, convert to int
62        addr = (addr[0], int(port[:-1]))
63        self.sock.connect(addr)
64
65        # Verify we can do the handshake.  If that works, we'll call it good.
66        self.do_handshake(self.sock)
67