Address SonarCloud + Codacy findings on PR #181

Round-up of every issue both scanners flagged on this branch:

Library code:
- Drop unused imports (NONCE_BYTES in host.py, dataclasses.field in
  file_transfer.py).
- Replace the 17-parameter RemoteDesktopHost.__init__ with an
  AudioCaptureConfig dataclass (S107). GUI and tests now pass
  audio_config=AudioCaptureConfig(enabled=True, ...) instead of five
  separate kwargs, taking the parameter list down to 13.
- Define module-level constants for repeated literals (S1192):
  _NOT_CONNECTED_MESSAGE in viewer.py, _OPEN_CLIPBOARD_FAILED in
  clipboard.py, _INVALID_TRANSFER_ID_MESSAGE in file_transfer.py.
- Refactor RemoteDesktopViewer._recv_loop into a per-message dispatch
  table (S3776) — cognitive complexity 47 -> well under 15.
- Float equality on host.py:638 sleep_for == 0.0 -> <= 0.0 (S1244).
- Drop redundant exception classes from except tuples whenever a
  superclass is already listed (S5713). ConnectionError, ssl.SSLError
  and TimeoutError all derive from OSError.
- ws_protocol.py: opposite-operator (S1940), reword 'commented-out'
  comment (S125), pass usedforsecurity=False on the SHA-1 used by the
  RFC 6455 handshake (Bandit B324 / Semgrep insecure-hash).
- audio.py: replace the bare 'pass' in PortAudio's callback isolation
  with an explicit return + nosec B110 annotation.
- All ssl.SSLContext(...) calls now set minimum_version = TLSv1_2
  (S4423). User-opt-in insecure flows for self-signed certs are
  marked NOSONAR S5527/S4830 with a brief reason instead of changing
  behaviour.

GUI:
- Drop unused imports (os, QClipboard, QApplication, send_file).
- Extract a _scroll_amount(angle_delta) helper to flatten the nested
  ternary on _FrameDisplay.wheelEvent (S3358).

Tests:
- Optional[_FakeStream] type hints (S5890); NOSONAR S100 on the two
  PascalCase mock methods that mirror the sounddevice API.
- Replace bare 'pass' on the failure-stub stop() with an explanatory
  return (S1186).
- NOSONAR S5655 on intentional bad-type tests for encode_text and
  dispatch_input.
- Rename the unused 'tid' tuple element to '_tid' (S1481).
- flow_control test: assert len + value before isinstance check so
  Sonar's flow analysis can prove seen[0] is safe (S6466).

Behaviour is unchanged; tests still 295 pass on Windows.

Author: JE-Chen

je_auto_control/gui/remote_desktop_tab.py

@@ -13,19 +13,18 @@
   original remote-screen pixel space using the latest received frame's
   size.
 """
-import os
 import secrets
 import ssl
 from pathlib import Path
 from typing import Optional
 
 from PySide6.QtCore import QPoint, QRect, Qt, QThread, QTimer, Signal
 from PySide6.QtGui import (
-    QClipboard, QDragEnterEvent, QDropEvent, QGuiApplication, QImage,
+    QDragEnterEvent, QDropEvent, QGuiApplication, QImage,
     QKeyEvent, QMouseEvent, QPainter, QWheelEvent,
 )
 from PySide6.QtWidgets import (
-    QApplication, QCheckBox, QComboBox, QFileDialog, QGroupBox, QHBoxLayout,
+    QCheckBox, QComboBox, QFileDialog, QGroupBox, QHBoxLayout,
     QInputDialog, QLabel, QLineEdit, QMessageBox, QProgressBar, QPushButton,
     QSizePolicy, QSpinBox, QTabWidget, QVBoxLayout, QWidget,
 )
@@ -39,9 +38,9 @@
     WebSocketDesktopHost, WebSocketDesktopViewer,
 )
 from je_auto_control.utils.remote_desktop.audio import (
-    AudioBackendError, AudioPlayer, is_audio_backend_available,
+    AudioBackendError, AudioCaptureConfig, AudioPlayer,
+    is_audio_backend_available,
 )
-from je_auto_control.utils.remote_desktop.file_transfer import send_file
 from je_auto_control.utils.remote_desktop.host_id import (
     HostIdError, format_host_id, parse_host_id,
 )
@@ -102,6 +101,15 @@ def _key_event_to_ac(event: QKeyEvent) -> Optional[str]:
     return None
 
 
+def _scroll_amount(angle_delta: int) -> int:
+    """Return ``+1`` / ``-1`` / ``0`` for a Qt wheel ``angleDelta`` value."""
+    if angle_delta > 0:
+        return 1
+    if angle_delta < 0:
+        return -1
+    return 0
+
+
 class _FrameDisplay(QWidget):
     """Paints the latest frame and emits remapped input events.
 
@@ -209,8 +217,7 @@ def wheelEvent(self, event: QWheelEvent) -> None:  # noqa: N802
         coords = self._to_remote(event.position().toPoint())
         if coords is None:
             return
-        delta = event.angleDelta().y()
-        amount = 1 if delta > 0 else -1 if delta < 0 else 0
+        amount = _scroll_amount(event.angleDelta().y())
         if amount:
             self.mouse_scrolled.emit(coords[0], coords[1], amount)
 
@@ -433,6 +440,7 @@ def _build_ssl_context(self) -> Optional[ssl.SSLContext]:
         if not cert_path or not key_path:
             raise ValueError(_t("rd_tls_both_required"))
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
         ctx.load_cert_chain(certfile=cert_path, keyfile=key_path)
         return ctx
 
@@ -443,7 +451,7 @@ def _start(self) -> None:
             token = self._token.text().strip()
         try:
             ssl_context = self._build_ssl_context()
-        except (OSError, ssl.SSLError, ValueError) as error:
+        except (OSError, ValueError) as error:
             QMessageBox.warning(self, _t("rd_host_start"), str(error))
             return
         host_cls = (WebSocketDesktopHost
@@ -459,8 +467,10 @@ def _start(self) -> None:
                 fps=float(self._fps.value()),
                 quality=self._quality.value(),
                 ssl_context=ssl_context,
-                enable_audio=self._enable_audio.isChecked()
-                and self._enable_audio.isEnabled(),
+                audio_config=AudioCaptureConfig(
+                    enabled=self._enable_audio.isChecked()
+                    and self._enable_audio.isEnabled(),
+                ),
             )
             host.start()
         except (OSError, ValueError, RuntimeError, AudioBackendError) as error:
@@ -686,7 +696,7 @@ def _connect(self) -> None:
         except AuthenticationError as error:
             QMessageBox.warning(self, _t("rd_viewer_connect"), str(error))
             return
-        except (OSError, ConnectionError, RuntimeError, ssl.SSLError) as error:
+        except (OSError, RuntimeError) as error:
             QMessageBox.warning(self, _t("rd_viewer_connect"), str(error))
             return
         registry._viewer = viewer  # noqa: SLF001  centralised lifecycle ownership
@@ -705,7 +715,9 @@ def _build_client_ssl_context(
         if transport not in ("TLS", "WSS"):
             return None
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
         if self._tls_insecure.isChecked():
+            # NOSONAR S5527 S4830  # reason: explicit user opt-in for self-signed
             ctx.check_hostname = False
             ctx.verify_mode = ssl.CERT_NONE
         else:
@@ -827,7 +839,7 @@ def _send(self, action: dict) -> None:
             return
         try:
             viewer.send_input(action)
-        except (OSError, ConnectionError) as error:
+        except OSError as error:
             self._error_signal.emit(str(error))
 
     def _send_mouse_move(self, x: int, y: int) -> None:
@@ -859,7 +871,7 @@ def _push_clipboard_to_host(self) -> None:
             return
         try:
             viewer.send_clipboard_text(text)
-        except (OSError, ConnectionError) as error:
+        except OSError as error:
             QMessageBox.warning(self, _t("rd_viewer_push_clipboard"),
                                 str(error))
             return
@@ -925,7 +937,7 @@ def relay(transfer_id, done, total):
             result = self._viewer.send_file(
                 self._source, self._dest, on_progress=relay,
             )
-        except (OSError, ConnectionError, RuntimeError) as error:
+        except (OSError, RuntimeError) as error:
             self.completed.emit("", False, str(error), self._dest)
             return
         self.completed.emit(

je_auto_control/utils/clipboard/clipboard.py

@@ -16,6 +16,8 @@
 from io import BytesIO
 from typing import Optional
 
+_OPEN_CLIPBOARD_FAILED = "OpenClipboard failed"
+
 
 def get_clipboard() -> str:
     """Return the current clipboard text (empty string if empty)."""
@@ -83,7 +85,7 @@ def _win_get() -> str:
     kernel32.GlobalUnlock.argtypes = [wintypes.HGLOBAL]
 
     if not user32.OpenClipboard(None):
-        raise RuntimeError("OpenClipboard failed")
+        raise RuntimeError(_OPEN_CLIPBOARD_FAILED)
     try:
         handle = user32.GetClipboardData(cf_unicodetext)
         if not handle:
@@ -131,7 +133,7 @@ def _win_set(text: str) -> None:
     ctypes.memmove(pointer, ctypes.addressof(data), size)  # NOSONAR S5655 false positive — Array is accepted by addressof
     kernel32.GlobalUnlock(handle)
     if not user32.OpenClipboard(None):
-        raise RuntimeError("OpenClipboard failed")
+        raise RuntimeError(_OPEN_CLIPBOARD_FAILED)
     try:
         user32.EmptyClipboard()
         if not user32.SetClipboardData(cf_unicodetext, handle):
@@ -256,7 +258,7 @@ def _win_set_image(png_bytes: bytes) -> None:
     ctypes.memmove(pointer, dib, len(dib))
     kernel32.GlobalUnlock(handle)
     if not user32.OpenClipboard(None):
-        raise RuntimeError("OpenClipboard failed")
+        raise RuntimeError(_OPEN_CLIPBOARD_FAILED)
     try:
         user32.EmptyClipboard()
         if not user32.SetClipboardData(cf_dib, handle):

je_auto_control/utils/remote_desktop/audio.py

@@ -11,6 +11,7 @@
 without noticeably starving the video pipe.
 """
 import threading
+from dataclasses import dataclass
 from typing import Callable, Optional
 
 DEFAULT_SAMPLE_RATE = 16_000
@@ -47,6 +48,17 @@ def is_audio_backend_available() -> bool:
     return True
 
 
+@dataclass(frozen=True)
+class AudioCaptureConfig:
+    """Bundled tuning knobs for :class:`RemoteDesktopHost` audio capture."""
+
+    enabled: bool = False
+    device: Optional[int] = None
+    sample_rate: int = DEFAULT_SAMPLE_RATE
+    channels: int = DEFAULT_CHANNELS
+    block_frames: int = DEFAULT_BLOCK_FRAMES
+
+
 class AudioCapture:
     """Capture mono int16 PCM blocks and hand them to ``on_block`` as bytes.
 
@@ -125,9 +137,9 @@ def _raw_callback(self, indata, frames, time_info, status) -> None:
             return
         try:
             self._on_block(bytes(indata))
-        except Exception:  # noqa: BLE001  callback isolation
+        except Exception:  # noqa: BLE001  callback isolation  # nosec B110  # reason: PortAudio callback must never raise
             # We must not propagate user callback errors back into PortAudio.
-            pass
+            return
 
 
 class AudioPlayer:

je_auto_control/utils/remote_desktop/file_transfer.py

@@ -20,7 +20,7 @@
 import os
 import threading
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, Dict, Optional, Tuple
 
@@ -30,6 +30,10 @@
 DEFAULT_CHUNK_SIZE = 256 * 1024
 TRANSFER_ID_LEN = 36  # str(uuid.uuid4()) length
 
+_INVALID_TRANSFER_ID_MESSAGE = (
+    f"transfer_id must be a {TRANSFER_ID_LEN}-char UUID string"
+)
+
 ProgressCallback = Callable[[str, int, int], None]
 CompleteCallback = Callable[[str, bool, Optional[str], str], None]
 
@@ -45,7 +49,7 @@ def new_transfer_id() -> str:
 
 def encode_begin(transfer_id: str, dest_path: str, size: int) -> bytes:
     if len(transfer_id) != TRANSFER_ID_LEN:
-        raise FileTransferError("transfer_id must be a 36-char UUID string")
+        raise FileTransferError(_INVALID_TRANSFER_ID_MESSAGE)
     return json.dumps({
         "transfer_id": transfer_id,
         "dest_path": str(dest_path),
@@ -70,7 +74,7 @@ def decode_begin(payload: bytes) -> Tuple[str, str, int]:
 
 def encode_chunk(transfer_id: str, chunk: bytes) -> bytes:
     if len(transfer_id) != TRANSFER_ID_LEN:
-        raise FileTransferError("transfer_id must be a 36-char UUID string")
+        raise FileTransferError(_INVALID_TRANSFER_ID_MESSAGE)
     return transfer_id.encode("ascii") + bytes(chunk)
 
 
@@ -84,7 +88,7 @@ def decode_chunk(payload: bytes) -> Tuple[str, bytes]:
 def encode_end(transfer_id: str, status: str = "ok",
                error: Optional[str] = None) -> bytes:
     if len(transfer_id) != TRANSFER_ID_LEN:
-        raise FileTransferError("transfer_id must be a 36-char UUID string")
+        raise FileTransferError(_INVALID_TRANSFER_ID_MESSAGE)
     body: Dict[str, Any] = {"transfer_id": transfer_id, "status": status}
     if error is not None:
         body["error"] = str(error)
@@ -257,7 +261,7 @@ def send_file(channel, source_path: str, dest_path: str,
                 bytes_sent += len(chunk)
                 if on_progress is not None:
                     on_progress(transfer_id, bytes_sent, total_size)
-    except (OSError, ConnectionError) as error:
+    except OSError as error:
         channel.send_typed(
             MessageType.FILE_END,
             encode_end(transfer_id, status="error", error=str(error)),