[lldb] Add type hints to gdbclientutils.py

[dsandersllvm]

Everything in this should be python 3.9. The docs say the minimum is 3.8 but there's existing code in this suite that needs 3.9 so I think 3.9 is ok.

Issues:
qEcho() is passed an argument by the callers that the function didn't have Several functions in the base class would silently do nothing if not overriden. These now use @AbstractMethod to require overrides sendall() had inconsistent return types between overrides

[llvmbot]

@llvm/pr-subscribers-lldb

Author: Daniel Sanders (dsandersllvm)

Changes

Everything in this should be python 3.9. The docs say the minimum is 3.8 but there's existing code in this suite that needs 3.9 so I think 3.9 is ok.

Issues:
qEcho() is passed an argument by the callers that the function didn't have Several functions in the base class would silently do nothing if not overriden. These now use @abstractmethod to require overrides sendall() had inconsistent return types between overrides

---

Full diff: https://github.com/llvm/llvm-project/pull/162172.diff

1 Files Affected:

• (modified) lldb/packages/Python/lldbsuite/test/gdbclientutils.py (+43-27)

diff --git a/lldb/packages/Python/lldbsuite/test/gdbclientutils.py b/lldb/packages/Python/lldbsuite/test/gdbclientutils.py
index b603c35c8df09..5fe1bc3155386 100644
--- a/lldb/packages/Python/lldbsuite/test/gdbclientutils.py
+++ b/lldb/packages/Python/lldbsuite/test/gdbclientutils.py
@@ -1,3 +1,4 @@
+from abc import ABC, abstractmethod
 import ctypes
 import errno
 import io
@@ -5,6 +6,7 @@
 import socket
 import traceback
 from lldbsuite.support import seven
+from typing import Optional
 
 
 def checksum(message):
@@ -86,8 +88,8 @@ class MockGDBServerResponder:
     handles any packet not recognized in the common packet handling code.
     """
 
-    registerCount = 40
-    packetLog = None
+    registerCount: int = 40
+    packetLog: Optional[list[str]] = None
 
     class RESPONSE_DISCONNECT:
         pass
@@ -103,6 +105,7 @@ def respond(self, packet):
         Return the unframed packet data that the server should issue in response
         to the given packet received from the client.
         """
+        assert self.packetLog is not None
         self.packetLog.append(packet)
         if packet is MockGDBServer.PACKET_INTERRUPT:
             return self.interrupt()
@@ -242,7 +245,7 @@ def qProcessInfo(self):
     def qHostInfo(self):
         return "ptrsize:8;endian:little;"
 
-    def qEcho(self):
+    def qEcho(self, _: int):
         return "E04"
 
     def qQueryGDBServer(self):
@@ -263,10 +266,10 @@ def A(self, packet):
     def D(self, packet):
         return "OK"
 
-    def readRegisters(self):
+    def readRegisters(self) -> str:
         return "00000000" * self.registerCount
 
-    def readRegister(self, register):
+    def readRegister(self, register: int) -> str:
         return "00000000"
 
     def writeRegisters(self, registers_hex):
@@ -306,7 +309,8 @@ def haltReason(self):
         # SIGINT is 2, return type is 2 digit hex string
         return "S02"
 
-    def qXferRead(self, obj, annex, offset, length):
+    def qXferRead(self, obj: str, annex: str, offset: int,
+                  length: int) -> tuple[str | None, bool]:
         return None, False
 
     def _qXferResponse(self, data, has_more):
@@ -374,15 +378,17 @@ class UnexpectedPacketException(Exception):
         pass
 
 
-class ServerChannel:
+class ServerChannel(ABC):
     """
     A wrapper class for TCP or pty-based server.
     """
 
-    def get_connect_address(self):
+    @abstractmethod
+    def get_connect_address(self) -> str:
         """Get address for the client to connect to."""
 
-    def get_connect_url(self):
+    @abstractmethod
+    def get_connect_url(self) -> str:
         """Get URL suitable for process connect command."""
 
     def close_server(self):
@@ -394,10 +400,12 @@ def accept(self):
     def close_connection(self):
         """Close all resources used by the accepted connection."""
 
-    def recv(self):
+    @abstractmethod
+    def recv(self) -> bytes:
         """Receive a data packet from the connected client."""
 
-    def sendall(self, data):
+    @abstractmethod
+    def sendall(self, data: bytes) -> None:
         """Send the data to the connected client."""
 
 
@@ -428,11 +436,11 @@ def close_connection(self):
         self._connection.close()
         self._connection = None
 
-    def recv(self):
+    def recv(self) -> bytes:
         assert self._connection is not None
         return self._connection.recv(4096)
 
-    def sendall(self, data):
+    def sendall(self, data: bytes) -> None:
         assert self._connection is not None
         return self._connection.sendall(data)
 
@@ -444,10 +452,10 @@ def __init__(self):
         )[0]
         super().__init__(family, type, proto, addr)
 
-    def get_connect_address(self):
+    def get_connect_address(self) -> str:
         return "[{}]:{}".format(*self._server_socket.getsockname())
 
-    def get_connect_url(self):
+    def get_connect_url(self) -> str:
         return "connect://" + self.get_connect_address()
 
 
@@ -455,10 +463,10 @@ class UnixServerSocket(ServerSocket):
     def __init__(self, addr):
         super().__init__(socket.AF_UNIX, socket.SOCK_STREAM, 0, addr)
 
-    def get_connect_address(self):
+    def get_connect_address(self) -> str:
         return self._server_socket.getsockname()
 
-    def get_connect_url(self):
+    def get_connect_url(self) -> str:
         return "unix-connect://" + self.get_connect_address()
 
 
@@ -472,7 +480,7 @@ def __init__(self):
         self._primary = io.FileIO(primary, "r+b")
         self._secondary = io.FileIO(secondary, "r+b")
 
-    def get_connect_address(self):
+    def get_connect_address(self) -> str:
         libc = ctypes.CDLL(None)
         libc.ptsname.argtypes = (ctypes.c_int,)
         libc.ptsname.restype = ctypes.c_char_p
@@ -485,7 +493,7 @@ def close_server(self):
         self._secondary.close()
         self._primary.close()
 
-    def recv(self):
+    def recv(self) -> bytes:
         try:
             return self._primary.read(4096)
         except OSError as e:
@@ -494,8 +502,8 @@ def recv(self):
                 return b""
             raise
 
-    def sendall(self, data):
-        return self._primary.write(data)
+    def sendall(self, data: bytes) -> None:
+        self._primary.write(data)
 
 
 class MockGDBServer:
@@ -528,18 +536,21 @@ def stop(self):
             self._thread.join()
             self._thread = None
 
-    def get_connect_address(self):
+    def get_connect_address(self) -> str:
+        assert self._socket is not None
         return self._socket.get_connect_address()
 
-    def get_connect_url(self):
+    def get_connect_url(self) -> str:
+        assert self._socket is not None
         return self._socket.get_connect_url()
 
     def run(self):
+        assert self._socket is not None
         # For testing purposes, we only need to worry about one client
         # connecting just one time.
         try:
             self._socket.accept()
-        except:
+        except Exception:
             traceback.print_exc()
             return
         self._shouldSendAck = True
@@ -554,7 +565,7 @@ def run(self):
                 self._receive(data)
         except self.TerminateConnectionException:
             pass
-        except Exception as e:
+        except Exception:
             print(
                 "An exception happened when receiving the response from the gdb server. Closing the client..."
             )
@@ -587,7 +598,9 @@ def _parsePacket(self):
         Once a complete packet is found at the front of self._receivedData,
         its data is removed form self._receivedData.
         """
+        assert self._receivedData is not None
         data = self._receivedData
+        assert self._receivedDataOffset is not None
         i = self._receivedDataOffset
         data_len = len(data)
         if data_len == 0:
@@ -640,10 +653,13 @@ def _parsePacket(self):
         self._receivedDataOffset = 0
         return packet
 
-    def _sendPacket(self, packet):
-        self._socket.sendall(seven.bitcast_to_bytes(frame_packet(packet)))
+    def _sendPacket(self, packet: str):
+        assert self._socket is not None
+        framed_packet = seven.bitcast_to_bytes(frame_packet(packet))
+        self._socket.sendall(framed_packet)
 
     def _handlePacket(self, packet):
+        assert self._socket is not None
         if packet is self.PACKET_ACK:
             # Ignore ACKs from the client. For the future, we can consider
             # adding validation code to make sure the client only sends ACKs

[github-actions]

⚠️ Python code formatter, darker found issues in your code. ⚠️

You can test this locally with the following command:

darker --check --diff -r origin/main...HEAD lldb/packages/Python/lldbsuite/test/gdbclientutils.py

⚠️
The reproduction instructions above might return results for more than one PR
in a stack if you are using a stacked PR workflow. You can limit the results by
changing origin/main to the base branch/commit you want to compare against.
⚠️

View the diff from darker here.

--- gdbclientutils.py	2025-10-04 03:05:11.000000 +0000
+++ gdbclientutils.py	2025-10-06 22:10:17.465481 +0000
@@ -307,12 +307,13 @@
 
     def haltReason(self):
         # SIGINT is 2, return type is 2 digit hex string
         return "S02"
 
-    def qXferRead(self, obj: str, annex: str, offset: int,
-                  length: int) -> tuple[str | None, bool]:
+    def qXferRead(
+        self, obj: str, annex: str, offset: int, length: int
+    ) -> tuple[str | None, bool]:
         return None, False
 
     def _qXferResponse(self, data, has_more):
         return "%s%s" % ("m" if has_more else "l", escape_binary(data))