[lldb] Add type hints to gdbclientutils.py and use abstract base class (llvm#162172)

Everything in this commit should be python 3.8 compatible which has
required using older styles of type hints (e.g. `Optional[T]` rather
than 3.10's `T | None` and `List[T]` rather than 3.9's `list[T]`. There
are some python 3.9 type hints in other files which have not been
changed by this commit.

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

Compatibility was checked with:
```
uvx vermin -t 3.8 $(find lldb/packages/Python -name '*.py')
```
Compability of the type hints was checked with:
```
uvx vermin -t 3.8 --eval-annotations $(find lldb/packages/Python -name '*.py')
```
and type hint correctness was checked with
```
uvx pyright lldb/packages/Python/lldbsuite/test/gdbclientutils.py
```

Author: dsandersllvm

lldb/packages/Python/lldbsuite/test/gdbclientutils.py

@@ -1,10 +1,12 @@
+from abc import ABC, abstractmethod
 import ctypes
 import errno
 import io
 import threading
 import socket
 import traceback
 from lldbsuite.support import seven
+from typing import Optional, List, Tuple
 
 
 def checksum(message):
@@ -86,7 +88,7 @@ class MockGDBServerResponder:
     handles any packet not recognized in the common packet handling code.
     """
 
-    registerCount = 40
+    registerCount: int = 40
 
     class RESPONSE_DISCONNECT:
         pass
@@ -95,7 +97,7 @@ class RESPONSE_NONE:
         pass
 
     def __init__(self):
-        self.packetLog = []
+        self.packetLog: List[str] = []
 
     def respond(self, packet):
         """
@@ -241,7 +243,7 @@ def qProcessInfo(self):
     def qHostInfo(self):
         return "ptrsize:8;endian:little;"
 
-    def qEcho(self):
+    def qEcho(self, num: int):
         return "E04"
 
     def qQueryGDBServer(self):
@@ -262,10 +264,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):
@@ -305,7 +307,9 @@ 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[Optional[str], bool]:
         return None, False
 
     def _qXferResponse(self, data, has_more):
@@ -373,15 +377,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):
@@ -393,10 +399,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."""
 
 
@@ -427,11 +435,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)
 
@@ -443,21 +451,21 @@ 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()
 
 
 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()
 
 
@@ -471,7 +479,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
@@ -484,7 +492,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:
@@ -493,8 +501,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:
@@ -527,18 +535,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
@@ -553,7 +564,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..."
             )
@@ -586,7 +597,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:
@@ -639,10 +652,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