Index: lldb/test/API/functionalities/gdb_remote_client/TestPty.py =================================================================== --- /dev/null +++ lldb/test/API/functionalities/gdb_remote_client/TestPty.py @@ -0,0 +1,32 @@ +import lldb +from lldbsuite.test.lldbtest import * +from lldbsuite.test.decorators import * +from gdbclientutils import * + + +class TestPty(TestBase): + mydir = TestBase.compute_mydir(__file__) + + def setUp(self): + super().setUp() + self.server = MockPtyGDBServer() + self.server.start() + + def tearDown(self): + if self.process() is not None: + self.process().Kill() + self.server.stop() + super().tearDown() + + @skipIfWindows + def test_process_connect_sync(self): + """Test the gdb-remote command in synchronous mode""" + try: + self.dbg.SetAsync(False) + self.expect("platform select remote-gdb-server", + substrs=['Platform: remote-gdb-server', 'Connected: no']) + self.expect("process connect file://" + + self.server.get_connect_address(), + substrs=['Process', 'stopped']) + finally: + self.dbg.GetSelectedPlatform().DisconnectRemote() Index: lldb/test/API/functionalities/gdb_remote_client/gdbclientutils.py =================================================================== --- lldb/test/API/functionalities/gdb_remote_client/gdbclientutils.py +++ lldb/test/API/functionalities/gdb_remote_client/gdbclientutils.py @@ -1,8 +1,12 @@ +import ctypes import errno +import io import os import os.path +import pty import threading import socket +import tty import lldb import binascii import traceback @@ -352,21 +356,26 @@ def __init__(self): self.responder = MockGDBServerResponder() - def start(self): + def _bind_and_listen(self): family, type, proto, _, addr = socket.getaddrinfo("localhost", 0, proto=socket.IPPROTO_TCP)[0] - self._socket = socket.socket(family, type, proto) - + sock = socket.socket(family, type, proto) - self._socket.bind(addr) - self._socket.listen(1) + sock.bind(addr) + sock.listen(1) + return sock + def start(self): + self._socket = self._bind_and_listen() # Start a thread that waits for a client connection. self._thread = threading.Thread(target=self._run) self._thread.start() - def stop(self): + def _close(self): self._socket.close() + + def stop(self): + self._close() self._thread.join() self._thread = None @@ -510,6 +519,51 @@ class InvalidPacketException(Exception): pass + +class PtySocket(io.FileIO): + """Pseudo-pocket wrapping a pty.""" + + def __init__(self, fd): + tty.setraw(fd) + super().__init__(fd, 'r+') + + def accept(self): + return self, '' + + def close(self): + # the socket is reused, so don't close it yet + pass + + def recv(self, size): + return self.read(size) + + def sendall(self, data): + return self.write(data) + + def settimeout(self, timeout): + pass + + +class MockPtyGDBServer(MockGDBServer): + """ + A variation of MockGDBServer that uses a pty instead of TCP. + """ + + def _bind_and_listen(self): + self._master, self._slave = pty.openpty() + return PtySocket(self._master) + + def _close(self): + os.close(self._slave) + os.close(self._master) + + def get_connect_address(self): + libc = ctypes.CDLL(None) + libc.ptsname.argtypes = (ctypes.c_int,) + libc.ptsname.restype = ctypes.c_char_p + return libc.ptsname(self._master).decode() + + class GDBRemoteTestBase(TestBase): """ Base class for GDB client tests.