fixed linting errors (#119)

* fixed linting errors

* typing error in backwards compatibility fix

* fixed missing requests typing

* fixed missing import

Author: MauriceHendrix

dev-requirements.txt

@@ -247,6 +247,7 @@ cryptography==44.0.2
     # via jwskate
 pycparser==2.22
     # via jwskate
+types-requests==2.32.0.20250328
 
 # The following packages are considered to be unsafe in a requirements file:
 # pip

src/s2python/__init__.py

@@ -1,4 +1,5 @@
-from importlib.metadata import PackageNotFoundError, version, sys  # pragma: no cover
+import sys  # pragma: no cover
+from importlib.metadata import PackageNotFoundError, version  # pragma: no cover
 
 try:
     # Change here if project is renamed and does not equal the package name
@@ -9,6 +10,5 @@
 finally:
     del version, PackageNotFoundError
 
-    from s2python.communication.s2_connection import S2Connection, AssetDetails
-    sys.modules['s2python.s2_connection'] = sys.modules.get('s2python.communication.s2_connection', None)
-    
+    from s2python.communication.s2_connection import S2Connection, AssetDetails   # pragma: no cover
+    sys.modules['s2python.s2_connection'] = sys.modules['s2python.communication.s2_connection'] # pragma: no cover

src/s2python/authorization/client.py

@@ -6,6 +6,7 @@
 import json
 import uuid
 import datetime
+import logging
 from typing import Dict, Optional, Tuple, Union, List, Any
 
 
@@ -27,6 +28,9 @@
 PAIRING_TIMEOUT = datetime.timedelta(minutes=5)
 KEY_ALGORITHM = "RSA-OAEP-256"
 
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("S2AbstractClient")
 
 class PairingDetails(BaseModel):
     """Contains all details from the pairing process."""
@@ -188,7 +192,7 @@ def request_pairing(self) -> PairingResponse:
             self.store_key_pair(public_key, private_key)
 
         # Create pairing request
-        print("Creating pairing request")
+        logger.info("Creating pairing request")
         pairing_request = PairingRequest(
             token=self.token,
             publicKey=self._public_key,
@@ -198,14 +202,14 @@ def request_pairing(self) -> PairingResponse:
         )
 
         # Make pairing request
-        print("Making pairing request")
+        logger.info("Making pairing request")
         status_code, response_text = self._make_https_request(
             url=self.pairing_uri,
             method="POST",
             data=pairing_request.model_dump(exclude_none=True),
             headers={"Content-Type": "application/json"},
         )
-        print(f"Pairing request response: {status_code} {response_text}")
+        logger.info('Pairing request response: %s %s', status_code, response_text)
 
         # Parse response
         if status_code != 200:
@@ -290,14 +294,12 @@ def request_connection(self) -> ConnectionDetails:
                     connection_details = ConnectionDetails.model_validate(
                         connection_data
                     )
-                    print(f"Updated relative WebSocket URI to absolute: {full_ws_url}")
+                    logger.info('Updated relative WebSocket URI to absolute: %s', full_ws_url)
                 except (ValueError, TypeError, KeyError) as e:
-                    print(f"Failed to update WebSocket URI: {e}")
+                    logger.info('Failed to update WebSocket URI: %s', e)
             else:
                 # Log a warning but don't modify the URI if we can't create a proper absolute URI
-                print(
-                    "Received relative WebSocket URI but pairing_uri is not available to create absolute URL"
-                )
+                logger.info('Received relative WebSocket URI but pairing_uri is not available to create absolute URL')
 
         # Store for later use
         self._connection_details = connection_details

src/s2python/authorization/default_client.py

@@ -8,6 +8,7 @@
 import base64
 import json
 import uuid
+import logging
 from typing import Dict, Optional, Tuple, Union, List, Any, Mapping
 
 import requests
@@ -27,6 +28,9 @@
     PairingDetails,
 )
 
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("S2DefaultClient")
 
 class S2DefaultClient(S2AbstractClient):
     """Default implementation of the S2AbstractClient using the requests library for HTTP
@@ -63,7 +67,7 @@ def generate_key_pair(self) -> Tuple[str, str]:
         Returns:
             Tuple[str, str]: (public_key, private_key) pair as PEM encoded strings
         """
-        print("Generating key pair")
+        logger.info("Generating key pair")
         self._key_pair = Jwk.generate_for_alg(KEY_ALGORITHM).with_kid_thumbprint()
         self._public_jwk = self._key_pair
         self._private_jwk = self._key_pair
@@ -82,7 +86,7 @@ def store_key_pair(self, public_key: str, private_key: str) -> None:
             public_key: PEM encoded public key
             private_key: PEM encoded private key
         """
-        print("Storing key pair")
+        logger.info("Storing key pair")
         self._public_key = public_key
         self._private_key = private_key
 
@@ -182,12 +186,12 @@ def solve_challenge(self, challenge: Optional[str] = None) -> str:
                     decrypted_challenge_str=decrypted_challenge_str,
                 )
 
-            print(f"Decrypted challenge: {decrypted_challenge_str}")
+            logger.info('Decrypted challenge: %s', decrypted_challenge_str)
             return decrypted_challenge_str
 
         except (ValueError, TypeError, KeyError, json.JSONDecodeError) as e:
             error_msg = f"Failed to solve challenge: {e}"
-            print(error_msg)
+            logger.info(error_msg)
             raise RuntimeError(error_msg) from e
 
     def establish_secure_connection(self) -> Dict[str, Any]:
@@ -219,12 +223,8 @@ def establish_secure_connection(self) -> Dict[str, Any]:
                 "Challenge solution not available. Call solve_challenge first."
             )
 
-        print(
-            f"Would establish WebSocket connection to {self._connection_details.connectionUri}"
-        )
-        print(
-            f"Using solved challenge: {self._pairing_details.decrypted_challenge_str}"
-        )
+        logger.info('Establishing WebSocket connection to %s,', self._connection_details.connectionUri)
+        logger.info('Using solved challenge: %s', self._pairing_details.decrypted_challenge_str)
 
         # Placeholder for the connection object
         self._ws_connection = {
@@ -240,6 +240,5 @@ def close_connection(self) -> None:
         """
         if self._ws_connection:
 
-            print("Would close WebSocket connection")
-            self._ws_connection.close()
+            logger.info("Would close WebSocket connection")
             self._ws_connection = None

src/s2python/communication/examples/example_pairing_frbc_rm.py

@@ -87,3 +87,4 @@
 
     except Exception as e:
         logger.error("Error during pairing process: %s", e)
+        raise e

src/s2python/communication/examples/mock_s2_server.py

@@ -3,9 +3,6 @@
 import json
 from typing import Any
 import uuid
-from urllib.parse import urlparse, parse_qs
-import ssl
-import threading
 import logging
 import random
 import string
@@ -39,14 +36,14 @@ def generate_token() -> str:
 
 
 class MockS2Handler(http.server.BaseHTTPRequestHandler):
-    def do_POST(self) -> None:
+    def do_POST(self) -> None: # pylint: disable=C0103
         content_length = int(self.headers.get("Content-Length", 0))
         post_data = self.rfile.read(content_length).decode("utf-8")
 
         try:
             request_json = json.loads(post_data)
-            logger.info(f"Received request at {self.path} ")
-            # logger.info(f"Request body: {request_json}")
+            logger.info('Received request at %s', self.path)
+            logger.debug('Request body: %s', request_json)
 
             if self.path == "/requestPairing":
                 # Handle pairing request
@@ -59,8 +56,8 @@ def do_POST(self) -> None:
                 else:
                     request_token_string = token_obj
 
-                logger.info(f"Extracted token: {request_token_string}")
-                logger.info(f"Expected token: {PAIRING_TOKEN}")
+                logger.info('Extracted token: %s', request_token_string)
+                logger.info('Expected token: %s', PAIRING_TOKEN)
 
                 if request_token_string == PAIRING_TOKEN:
                     self.send_response(200)
@@ -115,24 +112,25 @@ def do_POST(self) -> None:
                 self.send_header("Content-Type", "application/json")
                 self.end_headers()
                 self.wfile.write(json.dumps({"error": "Endpoint not found"}).encode())
-                logger.error(f"Unknown endpoint: {self.path}")
+                logger.error('Unknown endpoint: %s', self.path)
 
         except Exception as e:
             self.send_response(500)
             self.send_header("Content-Type", "application/json")
             self.end_headers()
             self.wfile.write(json.dumps({"error": str(e)}).encode())
-            logger.error(f"Error handling request: {e}")
+            logger.error('Error handling request: %s', e)
+            raise e
 
-    def log_message(self, format: str, *args: Any) -> None:
-        logger.info(format % args)
+    def log_message(self, format: str, *args: Any) -> None: # pylint: disable=W0622
+        logger.info(format % args) # pylint: disable=W1201
 
 
 def run_server() -> None:
     with socketserver.TCPServer(("localhost", HTTP_PORT), MockS2Handler) as httpd:
-        logger.info(f"Mock S2 Server running at http://localhost:{HTTP_PORT}")
-        logger.info(f"Use pairing token: {PAIRING_TOKEN}")
-        logger.info(f"Pairing endpoint: http://localhost:{HTTP_PORT}/requestPairing")
+        logger.info('Mock S2 Server running at: http://localhost:%s', HTTP_PORT)
+        logger.info('Use pairing token: %s', PAIRING_TOKEN)
+        logger.info('Pairing endpoint: http://localhost:%s/requestPairing', HTTP_PORT)
         httpd.serve_forever()
 
 

src/s2python/communication/examples/mock_s2_websocket.py

@@ -1,9 +1,9 @@
 import asyncio
-import websockets
 import logging
 import json
 import uuid
 from datetime import datetime, timezone
+import websockets
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -18,7 +18,7 @@ async def handle_connection(
     websocket: websockets.WebSocketServerProtocol, path: str
 ) -> None:
     client_id = str(uuid.uuid4())
-    logger.info(f"Client {client_id} connected on path: {path}")
+    logger.info('Client %s connected on path: %s', client_id, path)
 
     try:
         # Send handshake message to client
@@ -29,13 +29,13 @@ async def handle_connection(
             "timestamp": datetime.now(timezone.utc).isoformat(),
         }
         await websocket.send(json.dumps(handshake))
-        logger.info(f"Sent handshake to client {client_id}")
+        logger.info('Sent handshake to client %s', client_id)
 
         # Listen for messages
         async for message in websocket:
             try:
                 data = json.loads(message)
-                logger.info(f"Received message from client {client_id}: {data}")
+                logger.info('Received message from client %s: %s', client_id, data)
 
                 # Extract message type
                 message_type = data.get("type", "")
@@ -50,7 +50,7 @@ async def handle_connection(
                     "status": "OK",
                 }
                 await websocket.send(json.dumps(reception_status))
-                logger.info(f"Sent reception status for message {message_id}")
+                logger.info('Sent reception status for message %s', message_id)
 
                 # Handle specific message types
                 if message_type == "HandshakeResponse":
@@ -59,21 +59,23 @@ async def handle_connection(
                 # For FRBC messages, you could add specific handling here
 
             except json.JSONDecodeError:
-                logger.error(f"Invalid JSON received from client {client_id}")
+                logger.error('Invalid JSON received from client %s', client_id)
             except Exception as e:
-                logger.error(f"Error processing message from client {client_id}: {e}")
+                logger.error('Error processing message from client %s: %s', client_id, e)
+                raise e
 
     except websockets.exceptions.ConnectionClosed:
-        logger.info(f"Connection with client {client_id} closed")
+        logger.info('Connection with client %s closed', client_id)
     except Exception as e:
-        logger.error(f"Error with client {client_id}: {e}")
+        logger.error('Error with client %s: %s', client_id, e)
+        raise e
     finally:
-        logger.info(f"Client {client_id} disconnected")
+        logger.info('Client %s disconnected', client_id)
 
 
 async def start_server() -> None:
     server = await websockets.serve(handle_connection, "localhost", WS_PORT)
-    logger.info(f"WebSocket server started on ws://localhost:{WS_PORT}")
+    logger.info('WebSocket server started on ws://localhost:%s', WS_PORT)
 
     # Keep the server running
     await server.wait_closed()