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    @debugserver_test
15    def test_reverse_connect_debugserver(self):
16        self._reverse_connect()
17
18    @llgs_test
19    @skipIfRemote  # reverse connect is not a supported use case for now
20    def test_reverse_connect_llgs(self):
21        self._reverse_connect()
22
23    def _reverse_connect(self):
24        # Reverse connect is the default connection method.
25        self.connect_to_debug_monitor()
26        # Verify we can do the handshake.  If that works, we'll call it good.
27        self.do_handshake(self.sock)
28
29    @debugserver_test
30    @skipIfRemote
31    def test_named_pipe_debugserver(self):
32        self._named_pipe()
33
34    @llgs_test
35    @skipIfRemote
36    @skipIfWindows
37    def test_named_pipe_llgs(self):
38        self._named_pipe()
39
40    def _named_pipe(self):
41        family, type, proto, _, addr = socket.getaddrinfo(
42            self.stub_hostname, 0, proto=socket.IPPROTO_TCP)[0]
43        self.sock = socket.socket(family, type, proto)
44        self.sock.settimeout(self.DEFAULT_TIMEOUT)
45
46        self.addTearDownHook(lambda: self.sock.close())
47
48        named_pipe_path = self.getBuildArtifact("stub_port_number")
49
50        # Create the named pipe.
51        os.mkfifo(named_pipe_path)
52
53        # Open the read side of the pipe in non-blocking mode.  This will
54        # return right away, ready or not.
55        named_pipe_fd = os.open(named_pipe_path, os.O_RDONLY | os.O_NONBLOCK)
56
57        self.addTearDownHook(lambda: os.close(named_pipe_fd))
58
59        args = self.debug_monitor_extra_args
60        if lldb.remote_platform:
61            args += ["*:0"]
62        else:
63            args += ["localhost:0"]
64
65        args += ["--named-pipe", named_pipe_path]
66
67        server = self.spawnSubprocess(
68            self.debug_monitor_exe,
69            args,
70            install_remote=False)
71
72        (ready_readers, _, _) = select.select(
73            [named_pipe_fd], [], [], self.DEFAULT_TIMEOUT)
74        self.assertIsNotNone(
75            ready_readers,
76            "write side of pipe has not written anything - stub isn't writing to pipe.")
77        port = os.read(named_pipe_fd, 10)
78        # Trim null byte, convert to int
79        addr = (addr[0], int(port[:-1]))
80        self.sock.connect(addr)
81
82        # Verify we can do the handshake.  If that works, we'll call it good.
83        self.do_handshake(self.sock)
84