Extended test scripts to get 100 % coverage

examples/echoserver/__init__.py

@@ -56,7 +56,7 @@
 
 .. code:: bash
 
-   $ python server_ipv6.py
+   $ python server.py
    Slip server listening on localhost, port 59454
    Incoming connection from ('127.0.0.1', 59458)
    Raw data received: b'\\xc0hallo\\xc0'

examples/echoserver/client.py

@@ -20,7 +20,8 @@
 
 import sliplib
 
-if __name__ == "__main__":
+
+def main() -> None:
     if len(sys.argv) != 2:  # noqa: PLR2004
         print("Usage: python client.py <port>")
         sys.exit(1)
@@ -37,3 +38,7 @@
         sock.send_msg(b_message)
         b_reply = sock.recv_msg()
         print("Response:", b_reply)
+
+
+if __name__ == "__main__":
+    main()

examples/echoserver/server.py

@@ -13,7 +13,7 @@
 the raw data that is received and sent.
 The request handler prints the decoded message,
 and then reverses the order of the bytes in the encoded message
-(so ``abc`` becomes ``cab``),
+(so ``abc`` becomes ``cba``),
 and sends it back to the client.
 """
 
@@ -24,44 +24,34 @@
 import socket
 import sys
 from socketserver import TCPServer
-from typing import TYPE_CHECKING, Any
-
-if TYPE_CHECKING:
-    if sys.version_info >= (3, 12):
-        from collections.abc import Buffer
-    else:
-        from typing_extensions import Buffer
-
-from _socket import dup
-
-from sliplib import SlipRequestHandler
+from typing import Any
 
+from sliplib import SlipRequestHandler, SlipSocket
 
-class _ChattySocket(socket.socket):
-    """A socket subclass that prints the raw data that is received and sent."""
 
-    def __init__(self, sock: socket.socket) -> None:
-        fd = dup(sock.fileno())
-        super().__init__(sock.family, sock.type, sock.proto, fileno=fd)
-        super().settimeout(sock.gettimeout())
+class _ChattySocket(SlipSocket):
+    """A SlipSocket subclass that prints the raw data that is received and sent."""
 
-    def recv(self, chunksize: int, *args: Any) -> bytes:
-        data = super().recv(chunksize, *args)
+    def recv_bytes(self) -> bytes:
+        data = super().recv_bytes()
         print("Raw data received:", data)
         return data
 
-    def sendall(self, data: Buffer, *args: Any) -> None:
+    def send_bytes(self, data: bytes) -> None:
         print("Sending raw data:", data)
-        super().sendall(data, *args)
+        super().send_bytes(data)
 
 
 class SlipHandler(SlipRequestHandler):
     """A SlipRequestHandler that echoes the received message with the bytes in reversed order."""
 
-    def setup(self) -> None:
-        self.request = _ChattySocket(self.request)
-        print(f"Incoming connection from {self.request.getpeername()}")
-        super().setup()
+    def __init__(self, request: socket.socket | SlipSocket, *args: Any) -> None:
+        if isinstance(request, SlipSocket):
+            request.__class__ = _ChattySocket
+        else:
+            request = _ChattySocket(request)
+        print(f"Incoming connection from {request.getpeername()}")
+        super().__init__(request, *args)
 
     # Dedicated handler to show the encoded bytes.
     def handle(self) -> None:
@@ -81,10 +71,14 @@ class TCPServerIPv6(TCPServer):
     address_family = socket.AF_INET6
 
 
-if __name__ == "__main__":
+def main() -> None:
     if len(sys.argv) > 1 and sys.argv[1].lower() == "ipv6":
         server = TCPServerIPv6(("localhost", 0), SlipHandler)  # type: TCPServer
     else:
         server = TCPServer(("localhost", 0), SlipHandler)
     print("Slip server listening on localhost, port", server.server_address[1])
     server.handle_request()
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -66,7 +66,7 @@ cov = [
 python = "3.12"
 extra-dependencies = [
     "mypy>=1.0.0",
-    "typing_extensions"
+    "typing_extensions",
 ]
 
 [tool.hatch.envs.types.scripts]
@@ -105,8 +105,8 @@ omit = [
 ]
 
 [tool.coverage.paths]
-sliplib = ["src/sliplib", "*/SlipLib/src/sliplib"]
-tests = ["tests", "*/SlipLib/tests"]
+sliplib = ["src/sliplib"]
+tests = ["tests"]
 
 [tool.coverage.report]
 show_missing = true

src/sliplib/__init__.py

@@ -67,7 +67,7 @@
 .. automodule:: sliplib.slipwrapper
 .. automodule:: sliplib.slipstream
 .. automodule:: sliplib.slipsocket
-.. automodule:: sliplib.sliprequesthandler
+.. automodule:: sliplib.slipserver
 
 Exceptions
 ----------

src/sliplib/slipserver.py

@@ -9,9 +9,12 @@
 .. autoclass:: SlipRequestHandler
    :show-inheritance:
 
-   .. automethod:: setup
    .. automethod:: handle
    .. automethod:: finish
+
+.. autoclass:: SlipServer
+   :show-inheritance:
+
 """
 from __future__ import annotations
 
@@ -39,9 +42,9 @@ class SlipRequestHandler(BaseRequestHandler):
     def __init__(self, request: socket.socket | SlipSocket, client_address: TCPAddress, server: TCPServer):
         """Initializes the request handler.
 
-        The type of the :arg:`request` parameter depends on the type of server
+        The type of the :attr:`request` parameter depends on the type of server
         that instantiates the request handler.
-        If the server is a SlipServer, then :arg:`request` is a SlipSocket.
+        If the server is a SlipServer, then :attr:`request` is a SlipSocket.
         Otherwise, it is a regular socket, and must be wrapped in a SlipSocket
         before it can be used.
 
@@ -53,8 +56,15 @@ def __init__(self, request: socket.socket | SlipSocket, client_address: TCPAddre
                 The remote TCP addresss.
 
             server:
-                The TCPServer or SlipServer instance that instantiated this handler object.
+                The server instance that instantiated this handler object.
         """
+        if server.socket_type != socket.SOCK_STREAM:
+            message = (
+                f"{self.__class__.__name__} instance can only be used "
+                f"with a TCP server (got {server.__class__.__name__})"
+            )
+            raise TypeError(message)
+
         if not isinstance(request, SlipSocket):
             request = SlipSocket(request)
         super().__init__(cast(socket.socket, request), client_address, server)
@@ -88,23 +98,33 @@ def finish(self) -> None:
 
 
 class SlipServer(TCPServer):
-    """Base class for SlipSocket based server classes."""
+    """Base class for SlipSocket based server classes.
+
+    This is a convenience class, that offers a minor enhancement
+    over the regular :class:`TCPServer` from the standard library.
+    The class :class:`TCPServer` is hardcoded to use only IPv4 addresses. It must be subclassed
+    in order to use IPv6 addresses.
+    The :class:`SlipServer` class uses the address that is provided during instantiation
+    to determine if it must muse an IPv4 or IPv6 socket.
+    """
 
     def __init__(
         self,
         server_address: TCPAddress,
         handler_class: type[SlipRequestHandler],
         bind_and_activate: bool = True,  # noqa: FBT001 FBT002
     ):
+        """
+
+        Args:
+            server_address: The address on which the server listens
+            handler_class: The class that will be instantiated to handle an incoming request
+            bind_and_activate: Flag to indicate if the server must be bound and activated at creation time.
+        """
         if self._is_ipv6_address(server_address):
             self.address_family = socket.AF_INET6
         super().__init__(server_address[0:2], handler_class, bind_and_activate)
-
-    def server_bind(self) -> None:
-        """Make the server socket into a SLIP socket and bind it to the server address."""
-        if not isinstance(self.socket, SlipSocket):
-            self.socket = cast(socket.socket, SlipSocket(self.socket))
-        super().server_bind()
+        self.socket = cast(socket.socket, SlipSocket(self.socket))
 
     @staticmethod
     def _is_ipv6_address(server_address: TCPAddress) -> bool:

src/sliplib/slipsocket.py

@@ -121,7 +121,6 @@ class SlipSocket(SlipWrapper[socket.socket]):
     _chunk_size = 4096
 
     def __init__(self, sock: socket.SocketType):
-        # pylint: disable=missing-raises-doc
         """
         To instantiate a :class:`SlipSocket`, the user must provide
         a pre-constructed TCP `socket`.
@@ -222,6 +221,14 @@ def getsockopt(self, *args: Any) -> int | bytes:
         """
         return self.socket.getsockopt(*args)
 
+    def gettimeout(self) -> float | None:
+        """Get the socket option from the embedded socket.
+
+        Returns:
+            The integer or bytes representing the value of the socket option.
+        """
+        return self.socket.gettimeout()
+
     def listen(self, backlog: int | None = None) -> None:
         """Enable a `SlipSocket` server to accept connections.
 

src/sliplib/slipstream.py

@@ -14,6 +14,7 @@
    A :class:`SlipStream` instance has the following attributes in addition to the attributes
    offered by its base class :class:`SlipWrapper`:
 
+   .. autoattribute:: chunk_size
    .. autoattribute:: readable
    .. autoattribute:: writable
 """
@@ -99,7 +100,7 @@ def __init__(self, stream: IOStream, chunk_size: int = io.DEFAULT_BUFFER_SIZE):
         Args:
             stream: The byte stream that will be wrapped.
 
-            chunk_size: the number of bytes to read per read operation.
+            chunk_size: The number of bytes to read per read operation.
                 The default value for `chunck_size` is `io.DEFAULT_BUFFER_SIZE`.
 
                 Setting the `chunk_size` is useful when the stream has a low bandwidth
@@ -128,7 +129,10 @@ def __init__(self, stream: IOStream, chunk_size: int = io.DEFAULT_BUFFER_SIZE):
         if hasattr(stream, "encoding"):
             error_msg = f"{stream.__class__.__name__} object is not a byte stream"
             raise TypeError(error_msg)
-        self._chunk_size = chunk_size if chunk_size > 0 else io.DEFAULT_BUFFER_SIZE
+
+        #: The number of bytes to read during each read operation.
+        self.chunk_size = chunk_size if chunk_size > 0 else io.DEFAULT_BUFFER_SIZE
+
         super().__init__(stream)
 
     def send_bytes(self, packet: bytes) -> None:
@@ -139,7 +143,7 @@ def send_bytes(self, packet: bytes) -> None:
 
     def recv_bytes(self) -> bytes:
         """See base class"""
-        return b"" if self._stream_is_closed else self.stream.read(self._chunk_size)
+        return b"" if self._stream_is_closed else self.stream.read(self.chunk_size)
 
     @property
     def readable(self) -> bool:

tests/examples/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (c) 2024 Ruud de Jong
+# This file is part of the SlipLib project which is released under the MIT license.
+# See https://github.com/rhjdjong/SlipLib for details.

tests/examples/echoserver/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (c) 2024 Ruud de Jong
+# This file is part of the SlipLib project which is released under the MIT license.
+# See https://github.com/rhjdjong/SlipLib for details.