Merge pull request #32 from bp100a/31-update-mockusbip-server-to-use-the-sockewrapper-class-for-greater-testability

31 update mockusbip server to use the sockewrapper class for greater testability

Author: bp100a

CHANGES

@@ -1,6 +1,12 @@
 # CHANGES
+## [1.0.4] - 2024-09-06
+- for testability, MockUSBIP server uses the SocketWrapper class
+  for socket abstraction, will allow injecting errors for testing
+
 ## [1.0.3] - 2024-09-06
 - refactoring for better test coverage of the Mock USBIP server
+- for testability, MockUSBIP server uses the SocketWrapper class
+  for socket abstraction, will allow injecting errors for testing
 
 ## [1.0.2] - 2024-09-05
 - refactor some code to increase test coverage of the usbip_client.py

coverage.svg

@@ -48,7 +48,7 @@ Repository = "https://github.com/bp100a/serial-usbipclient"
 name = "serial-usbipclient"
 authors = ["Harry J. Collins <[email protected]>"]
 maintainers = ["Harry J. Collins <[email protected]>"]
-version = "1.0.3"
+version = "1.0.4"
 description = "Simple serial connectivity to CDC (serial) USB devices exposed by a USBIPD service."
 readme = "README.md"
 package-mode = true

pyproject.toml

@@ -1,5 +1,6 @@
 """wraps socket so we can abstract if for testing (dependency injection)"""
 from socket import AddressFamily, SocketKind, socket
+from typing import Any
 
 from typing_extensions import Buffer
 
@@ -46,3 +47,24 @@ def sendall(self, data: bytes) -> None:
     def recv(self, size: int) -> bytes:
         """read from our socket"""
         return self._socket.recv(size)
+
+    def bind(self, address: tuple[str, int]) -> None:
+        """bind to the socket"""
+        return self._socket.bind(address)
+
+    def listen(self, backlog: int):
+        """listen for connections"""
+        return self._socket.listen(backlog)
+
+    def accept(self) -> tuple[socket, Any]:
+        """accept a connection"""
+        return self._socket.accept()
+
+    def fileno(self) -> int:
+        """return the sockets file"""
+        return self._socket.fileno()
+
+    @property
+    def raw_socket(self) -> socket:
+        """get the underlying socket"""
+        return self._socket

serial_usbipclient/socket_wrapper.py

@@ -34,6 +34,7 @@
     UrbSetupPacket, URBStandardDeviceRequest)
 from serial_usbipclient.protocol.usb_descriptors import DescriptorType
 from serial_usbipclient.protocol.usbip_defs import BasicCommands, Direction
+from serial_usbipclient.socket_wrapper import SocketWrapper
 
 LOGGER: logging.Logger = logging.getLogger(__name__)
 
@@ -423,12 +424,13 @@ class DebugCommands(StrEnum):
         NO_WRITE_RESPONSE = 'no-write-response'  # write does not return acknowledgement
         NO_READ_RESPONSE = 'no-read-response'  # suppress an expected read
 
-    def __init__(self, host: str, port: int):
+    def __init__(self, host: str, port: int, socket_class: type = SocketWrapper):
         """set up our instance"""
         self.host: str = host
         self.port: int = port
         self.logger: logging.Logger = LOGGER
-        self.server_socket: Optional[socket.socket] = None
+        self.server_socket: Optional[SocketWrapper] = None
+        self._socket_class: type = socket_class
         self.thread: Optional[Thread] = Thread(name=f'mock-usbip@{self.host}:{self.port}', target=self.run_server, daemon=True)
         self.event: Event = Event()
         self._is_windows: bool = platform.system() == 'Windows'
@@ -676,7 +678,7 @@ def wait_for_message(self, conn: Optional[USBIPServerClient] = None) -> tuple[US
         if self._wakeup is None or self.server_socket is None:
             raise ValueError("neither the wakeup or server socket can be empty!")
 
-        rlist: list[socket.socket | USBIPServerClient] = [self._wakeup.listener, self.server_socket]
+        rlist: list[socket.socket | USBIPServerClient] = [self._wakeup.listener, self.server_socket.raw_socket]
         if conn:
             if conn not in self._clients:
                 self._clients.append(conn)
@@ -693,7 +695,7 @@ def wait_for_message(self, conn: Optional[USBIPServerClient] = None) -> tuple[US
                     if socket_read == self._wakeup.listener:  # time to bail
                         self.logger.info("Wakeup!")
                         raise OrderlyExit("wakeup!")
-                    elif socket_read == self.server_socket:  # someone is knocking
+                    elif socket_read == self.server_socket.raw_socket:  # someone is knocking
                         self.logger.info(f"wait_for_message(): accept() {self.server_socket=}")
                         new_conn, address = self.server_socket.accept()  # accept new connection
                         client: USBIPServerClient = USBIPServerClient(connection=new_conn, address=address)
@@ -750,12 +752,12 @@ def read_message(self, client: Optional[USBIPServerClient] = None) -> tuple[USBI
 
     def run_server(self):
         """standup the server, start listening"""
-        self.server_socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
-        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        self.server_socket.bind((self.host, self.port))
-        self.server_socket.settimeout(None)  # so our accept() will block
+        self.server_socket = self._socket_class(socket.AF_INET, socket.SOCK_STREAM)
+        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # type: ignore[union-attr]
+        self.server_socket.bind((self.host, self.port))  # type: ignore[union-attr]
+        self.server_socket.settimeout(None)  # type: ignore[union-attr]  # accept should block
 
-        self.server_socket.listen(1)  # only allow one connection (testing)
+        self.server_socket.listen(1)  # type: ignore[union-attr] # only allow one connection (testing)
         self.event.set()
         self.logger.info("\nmock USBIP server started @%s:%s", self.host, self.port)
         try: