Merge pull request #23 from bp100a/abstract-socket-for-testability

Abstract socket for testability

Author: bp100a

README.md

@@ -89,7 +89,7 @@ failing USBIPD attachments, and `1-2` & `1-3` have the same VID/PID (for testing
 During testing, the `MockUSBIP` service acts as a stand-in for an actual USBIP server. Since the tests are run in parallel, 
 the port on which the service listens must be unique for each unit test. This is accomplished by the `conftest.py::pytest_sessionstart` which is run
 when the pytest session is started and collects all tests being run into a file called `lists_of_tests.json`. The unit test's index into this array
-is used to determine the offset to be added to the port base (typically **3240**).
+is used to determine the offset to be added to the port base (typically **3240**). The `SocketWrapper` class is overridden in unit tests to inject low-level socket failures.
 
 
 Tooling
@@ -124,17 +124,9 @@ pip install -r tool_requirements.txt
 
 Tooling is not needed to run the package but is required for testing & packaging.
 
-Build Process
+Build Process (_now using Poetry!_)
 --------------------------------------------------------------------------
 ___
-Using `pip-compile` from the [pip-tools](https://pypi.org/project/pip-tools/) package, read the docs [here](https://pip-tools.readthedocs.io/en/latest/)
-
-```bash
-pip-compile requirements.in
-```
-
-Release to PyPi (_now using Poetry!_)
-----------------------------------------------------------------------------
 Store the PyPi API token
 ```bash
 keyring set https://upload.pypi.org/legacy/ __token__

coverage.svg

@@ -6,7 +6,6 @@ build-backend = "poetry.core.masonry.api"
 name = "serial-usbipclient"
 authors = [{name = "Harry J. Collins", email="bp100a@gmail.com"}]
 maintainers = [{name = "Harry J. Collins", email="bp100a@gmail.com"}]
-version = "0.0.7"
 description = "Simple serial connectivity to CDC (serial) USB devices exposed by a USBIPD service."
 readme = "README.md"
 license = {text = "BSD License"}
@@ -47,7 +46,7 @@ Repository = "https://github.com/bp100a/serial-usbipclient"
 name = "serial-usbipclient"
 authors = ["Harry J. Collins <bp100a@gmail.com>"]
 maintainers = ["Harry J. Collins <bp100a@gmail.com>"]
-version = "0.1.2"
+version = "0.1.3"
 description = "Simple serial connectivity to CDC (serial) USB devices exposed by a USBIPD service."
 readme = "README.md"
 package-mode = true

pyproject.toml

@@ -0,0 +1,52 @@
+"""wraps socket so we can abstract if for testing (dependency injection)"""
+from socket import AddressFamily, SocketKind, socket
+
+from typing_extensions import Buffer
+
+
+class SocketWrapper:
+    """wraps the raw socket class, exposes appropriate methods"""
+    SOCKET_TIMEOUT: float = 0.005
+    SERVER_CONNECT_TIMEOUT: float = 1.0
+
+    def __init__(self, family: AddressFamily, kind: SocketKind):
+        """set up socket"""
+        self._socket: socket = socket(family, kind)
+        self._address: tuple[str, int] = ('', 0)
+
+    def raw_socket(self) -> socket:
+        """return the underly, raw socket"""
+        return self._socket
+
+    def settimeout(self, timeout: float | None) -> None:
+        """set the timeout on the socket"""
+        self._socket.settimeout(timeout)
+
+    def connect(self, address: tuple[str, int]) -> None:
+        """connect to the remote host"""
+        self._address = address
+        self._socket.connect(self._address)
+
+    def setsockopt(self, level: int, option: int, value: int | Buffer) -> None:
+        """set the socket's options"""
+        self._socket.setsockopt(level, option, value)
+
+    def shutdown(self, how: int) -> None:
+        """shutdown the underlying socket"""
+        self._socket.shutdown(how)
+
+    def close(self) -> None:
+        """close the socket"""
+        self._socket.close()
+
+    def getsockname(self) -> tuple[str, int]:
+        """get the socket's name"""
+        return self._socket.getsockname()[1]  # always returning (address, port)
+
+    def sendall(self, data: bytes) -> None:
+        """send out to our socket"""
+        self._socket.sendall(data)
+
+    def recv(self, size: int) -> bytes:
+        """read from our socket"""
+        return self._socket.recv(size)

requirements.in

@@ -1 +1 @@
-["serial_usbipclient.tests.test_connection.py.testusbipconnection.test_connection", "serial_usbipclient.tests.test_connection.py.testusbipconnection.test_connection_shutdown", "serial_usbipclient.tests.test_connection.py.testusbipconnection.test_queue_urbs", "serial_usbipclient.tests.test_descriptors.py.testinterfacedescriptor.test_descriptors", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_mock_usbip_server_connection", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_mock_usbip_server_startup", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_mocked_response", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_reading_paths", "serial_usbipclient.tests.test_mock_usbip_server.py.testdeviceconfiguration.test_lsusb_parsing", "serial_usbipclient.tests.test_mock_usbip_server.py.testdeviceconfiguration.test_usbip_path", "serial_usbipclient.tests.test_packaging.py.testversion.test_version", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_cmd_response", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_cmd_submit", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_iso_packet_descriptor", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_metastruct_size", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_queue_urb", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_request_devlist", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_sizing", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_urb_endianness", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_configuration", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_delimited_read", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_failed_attachment", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_hardware_id_formatting", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_invalid_response", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_no_read_response", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_no_write_response", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_read_timeout", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_read_write", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_readline", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_readline_timeout", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_restore_connection", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_restore_unattachable_connection", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_restore_unknown_connection", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_timeout_error", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_unknown_device", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_write_string", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_configuration_descriptor", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_descriptor_handlers", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_endpoint_descriptor_handlers", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_generate_configuration", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_interface_association_handlers"]
+["serial_usbipclient.tests.test_connection.py.testusbipconnection.test_connection", "serial_usbipclient.tests.test_connection.py.testusbipconnection.test_connection_shutdown", "serial_usbipclient.tests.test_connection.py.testusbipconnection.test_queue_urbs", "serial_usbipclient.tests.test_descriptors.py.testinterfacedescriptor.test_descriptors", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_mock_usbip_server_connection", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_mock_usbip_server_startup", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_mocked_response", "serial_usbipclient.tests.test_mock_usbip_server.py.testmockusbipserver.test_reading_paths", "serial_usbipclient.tests.test_mock_usbip_server.py.testdeviceconfiguration.test_lsusb_parsing", "serial_usbipclient.tests.test_mock_usbip_server.py.testdeviceconfiguration.test_usbip_path", "serial_usbipclient.tests.test_packaging.py.testversion.test_version", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_cmd_response", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_cmd_submit", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_iso_packet_descriptor", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_metastruct_size", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_queue_urb", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_request_devlist", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_sizing", "serial_usbipclient.tests.test_packet_generation.py.testpacketgeneration.test_urb_endianness", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_configuration", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_delimited_read", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_failed_attachment", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_hardware_id_formatting", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_invalid_response", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_no_read_response", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_no_write_response", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_read_timeout", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_read_write", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_readline", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_readline_timeout", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_restore_connection", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_restore_unattachable_connection", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_restore_unknown_connection", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_timeout_error", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_unknown_device", "serial_usbipclient.tests.test_rw_device.py.testreadwrite.test_write_string", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_attachment", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_connection", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_gai_error", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_list_published", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_recv_error", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_restore_connection", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_restore_connection_no_device", "serial_usbipclient.tests.test_socket_handling.py.testsocketwrapper.test_timeout_error", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_configuration_descriptor", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_descriptor_handlers", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_endpoint_descriptor_handlers", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_generate_configuration", "serial_usbipclient.tests.test_urb_packets.py.testurbpackets.test_interface_association_handlers"]