Fix SonarCloud and Codacy findings

docs/source/conf.py

@@ -17,7 +17,7 @@
 # -- Project information -----------------------------------------------------
 
 project = "PyBreeze"
-copyright = "2020 ~ Now, JE-Chen"
+copyright = "2020 ~ Now, JE-Chen"  # noqa: A001 — Sphinx requires this exact variable name
 author = "JE-Chen"
 
 # -- General configuration ---------------------------------------------------

pybreeze/extend/process_executor/file_runner_process.py

@@ -75,7 +75,9 @@ def _compile_and_run(self, compiler: str, args: list, output_flag: str, file_pat
         self._append_text(f"[Compile] {' '.join(compile_cmd)}\n", is_error=False)
 
         try:
-            result = subprocess.run(
+            # Runs the plugin-configured compiler against a file the user opened.
+            # shell=False, bounded timeout. nosec B603.
+            result = subprocess.run(  # nosec B603  # nosemgrep  # noqa: S603
                 compile_cmd,
                 capture_output=True,
                 timeout=60,
@@ -107,7 +109,9 @@ def _start_process(self, command: list[str], cleanup_binary: str | None = None)
         self._append_text(f"> {cmd_display}\n", is_error=False)
 
         try:
-            self.process = subprocess.Popen(
+            # Run the user's plugin-configured command. shell=False is explicit;
+            # argv comes from a plugin run_config + user-opened file path. nosec B603.
+            self.process = subprocess.Popen(  # nosec B603  # nosemgrep  # noqa: S603
                 command,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,

pybreeze/extend/process_executor/python_task_process_manager.py

@@ -15,6 +15,7 @@
 from je_editor.pyside_ui.main_ui.save_settings.user_color_setting_file import actually_color_dict
 from je_editor.utils.venv_check.check_venv import check_and_choose_venv
 
+from pybreeze.extend.process_executor.queue_pump import pump_message_queue
 from pybreeze.pybreeze_ui.show_code_window.code_window import CodeWindow
 from pybreeze.utils.logging.logger import pybreeze_logger
 
@@ -82,7 +83,10 @@ def start_test_process(self, package: str, exec_str: str):
             "--execute_str",
             exec_str
         ]
-        self.process = subprocess.Popen(
+        # Launch user-authored automation script in a child interpreter.
+        # Argument list is validated upstream; shell=False, no user string ever
+        # reaches a shell. nosec B603 — intentional local process execution.
+        self.process = subprocess.Popen(  # nosec B603  # nosemgrep  # noqa: S603
             args,
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
@@ -121,28 +125,18 @@ def _append_text(self, text: str, is_error: bool = False) -> None:
 
     # Pyside UI update method
     def pull_text(self):
-        try:
-            if not self.run_output_queue.empty():
-                output_message = str(self.run_output_queue.get_nowait()).strip()
-                if output_message:
-                    self._append_text(output_message)
-            if not self.run_error_queue.empty():
-                error_message = str(self.run_error_queue.get_nowait()).strip()
-                if error_message:
-                    self._append_text(error_message, is_error=True)
-        except queue.Empty:
-            pass
-        if self.process is not None:
-            if self.process.returncode is not None:
-                if self.timer.isActive():
-                    self.timer.stop()
-                self.exit_program()
-            elif self.still_run_program:
-                # poll return code
-                self.process.poll()
-        else:
+        pump_message_queue(self.run_output_queue, self._append_text, is_error=False)
+        pump_message_queue(self.run_error_queue, self._append_text, is_error=True)
+        if self.process is None:
+            if self.timer.isActive():
+                self.timer.stop()
+            return
+        if self.process.returncode is not None:
             if self.timer.isActive():
                 self.timer.stop()
+            self.exit_program()
+        elif self.still_run_program:
+            self.process.poll()
 
     # exit program change run flag to false and clean read thread and queue and process
     def exit_program(self):

pybreeze/extend/process_executor/queue_pump.py

@@ -0,0 +1,33 @@
+"""Shared helper for the process-executor QTimer pump loop.
+
+The Python task runner and the Test Pioneer runner both drain the same
+``(output_queue, error_queue) → append_text`` pattern on a ~100 ms timer.
+This module factors the per-tick drain into a single function so changes
+(e.g., multi-message batching) can be made in one place.
+"""
+from __future__ import annotations
+
+import queue
+from collections.abc import Callable
+from queue import Queue
+
+
+def pump_message_queue(
+    q: Queue,
+    append_fn: Callable[[str, bool], None],
+    *,
+    is_error: bool,
+) -> None:
+    """Drain one pending message from *q* and forward it to *append_fn*.
+
+    Silently ignores ``queue.Empty`` (the ``empty()`` check is racy) and empty
+    strings after stripping.
+    """
+    try:
+        if q.empty():
+            return
+        message = str(q.get_nowait()).strip()
+        if message:
+            append_fn(message, is_error)
+    except queue.Empty:
+        pass

pybreeze/extend/process_executor/test_pioneer/test_pioneer_process_manager.py

@@ -2,9 +2,7 @@
 
 import queue
 import subprocess
-import sys
 import threading
-from pathlib import Path
 from queue import Queue
 from typing import TYPE_CHECKING
 
@@ -13,8 +11,9 @@
 from je_editor.pyside_ui.main_ui.save_settings.user_color_setting_file import actually_color_dict
 from je_editor.utils.venv_check.check_venv import check_and_choose_venv
 
-from pybreeze.pybreeze_ui.show_code_window.code_window import CodeWindow
 from pybreeze.extend.process_executor.python_task_process_manager import find_venv_path
+from pybreeze.extend.process_executor.queue_pump import pump_message_queue
+from pybreeze.pybreeze_ui.show_code_window.code_window import CodeWindow
 
 if TYPE_CHECKING:
     from pybreeze.pybreeze_ui.editor_main.main_ui import PyBreezeMainWindow
@@ -54,7 +53,10 @@ def __init__(
             "-e",
             executable_path
         ]
-        self._process: subprocess.Popen | None = subprocess.Popen(
+        # Launch the test_pioneer CLI in the user's configured Python interpreter.
+        # Argument list is assembled from a curated template + an executable path the
+        # user selected via file dialog. shell=False. nosec B603.
+        self._process: subprocess.Popen | None = subprocess.Popen(  # nosec B603  # nosemgrep  # noqa: S603
             args,
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
@@ -72,28 +74,18 @@ def _append_text(self, text: str, is_error: bool = False) -> None:
 
     # Pyside UI update method
     def pull_text(self):
-        try:
-            if not self._run_output_queue.empty():
-                output_message = str(self._run_output_queue.get_nowait()).strip()
-                if output_message:
-                    self._append_text(output_message)
-            if not self._run_error_queue.empty():
-                error_message = str(self._run_error_queue.get_nowait()).strip()
-                if error_message:
-                    self._append_text(error_message, is_error=True)
-        except queue.Empty:
-            pass
-        if self._process is not None:
-            if self._process.returncode is not None:
-                if self._timer.isActive():
-                    self._timer.stop()
-                self.exit_program()
-            elif self._still_run_program:
-                # poll return code
-                self._process.poll()
-        else:
+        pump_message_queue(self._run_output_queue, self._append_text, is_error=False)
+        pump_message_queue(self._run_error_queue, self._append_text, is_error=True)
+        if self._process is None:
+            if self._timer.isActive():
+                self._timer.stop()
+            return
+        if self._process.returncode is not None:
             if self._timer.isActive():
                 self._timer.stop()
+            self.exit_program()
+        elif self._still_run_program:
+            self._process.poll()
 
     # exit program change run flag to false and clean read thread and queue and process
     def exit_program(self):

pybreeze/extend_multi_language/extend_english.py

@@ -2,6 +2,9 @@
 
 from je_editor import english_word_dict
 
+_COT_PROMPT_EDITOR = "CoT Prompt Editor"
+_SKILL_PROMPT_EDITOR = "Skill Prompt Editor"
+
 # PyBreeze-specific English translations
 pybreeze_english_word_dict = {
     # application name
@@ -201,7 +204,7 @@
     "ai_code_review_gui_status_rejected": "[Rejected]",
     "ai_code_review_gui_status_save_failed": "Save failed",
     # CoT Prompt Editor
-    "cot_prompt_editor_window_title": "CoT Prompt Editor",
+    "cot_prompt_editor_window_title": _COT_PROMPT_EDITOR,
     "cot_prompt_editor_groupbox_edit_file_content": "Edit File Content",
     "cot_prompt_editor_button_create_file": "Create File",
     "cot_prompt_editor_button_save_file": "Save",
@@ -215,7 +218,7 @@
     "cot_prompt_editor_msgbox_no_file_selected": "No file selected",
     "cot_prompt_editor_file_not_exist": "(File {filename} does not exist)",
     # Skill Prompt Editor
-    "skill_prompt_editor_window_title": "Skill Prompt Editor",
+    "skill_prompt_editor_window_title": _SKILL_PROMPT_EDITOR,
     "skill_prompt_editor_groupbox_edit_file_content": "Edit File Content",
     "skill_prompt_editor_button_create_file": "Create File",
     "skill_prompt_editor_button_save_file": "Save",
@@ -236,10 +239,10 @@
     "extend_tools_menu_ssh_client_tab_label": "SSH Client",
     "extend_tools_menu_ai_code_review_tab_action": "AI Code-Review Tab",
     "extend_tools_menu_ai_code_review_tab_label": "AI Code-Review",
-    "extend_tools_menu_cot_prompt_editor_tab_action": "CoT Prompt Editor",
-    "extend_tools_menu_cot_prompt_editor_tab_label": "CoT Prompt Editor",
-    "extend_tools_menu_skill_prompt_editor_tab_action": "Skill Prompt Editor",
-    "extend_tools_menu_skill_prompt_editor_tab_label": "Skill Prompt Editor",
+    "extend_tools_menu_cot_prompt_editor_tab_action": _COT_PROMPT_EDITOR,
+    "extend_tools_menu_cot_prompt_editor_tab_label": _COT_PROMPT_EDITOR,
+    "extend_tools_menu_skill_prompt_editor_tab_action": _SKILL_PROMPT_EDITOR,
+    "extend_tools_menu_skill_prompt_editor_tab_label": _SKILL_PROMPT_EDITOR,
     "extend_tools_menu_skill_prompt_send_tab_label": "Skill Send GUI",
     "extend_tools_menu_dock_ssh_menu": "SSH",
     "extend_tools_menu_dock_ai_menu": "AI",

pybreeze/extend_multi_language/extend_traditional_chinese.py

@@ -2,6 +2,9 @@
 
 from je_editor import traditional_chinese_word_dict
 
+_COT_PROMPT_EDITOR = "CoT 提示詞編輯器"
+_SKILL_PROMPT_EDITOR = "Skill 提示詞編輯器"
+
 # PyBreeze-specific Traditional Chinese translations
 pybreeze_traditional_chinese_word_dict = {
     # application name
@@ -236,10 +239,10 @@
     "extend_tools_menu_ssh_client_tab_label": "SSH 用戶端",
     "extend_tools_menu_ai_code_review_tab_action": "AI 程式碼審查分頁",
     "extend_tools_menu_ai_code_review_tab_label": "AI 程式碼審查",
-    "extend_tools_menu_cot_prompt_editor_tab_action": "CoT 提示詞編輯器",
-    "extend_tools_menu_cot_prompt_editor_tab_label": "CoT 提示詞編輯器",
-    "extend_tools_menu_skill_prompt_editor_tab_action": "Skill 提示詞編輯器",
-    "extend_tools_menu_skill_prompt_editor_tab_label": "Skill 提示詞編輯器",
+    "extend_tools_menu_cot_prompt_editor_tab_action": _COT_PROMPT_EDITOR,
+    "extend_tools_menu_cot_prompt_editor_tab_label": _COT_PROMPT_EDITOR,
+    "extend_tools_menu_skill_prompt_editor_tab_action": _SKILL_PROMPT_EDITOR,
+    "extend_tools_menu_skill_prompt_editor_tab_label": _SKILL_PROMPT_EDITOR,
     "extend_tools_menu_skill_prompt_send_tab_label": "Skill 提示詞傳送 GUI",
     "extend_tools_menu_dock_ssh_menu": "SSH",
     "extend_tools_menu_dock_ai_menu": "AI",
@@ -249,8 +252,8 @@
     "extend_tools_menu_skill_prompt_editor_dock_action": "Skill 提示詞編輯器停駐窗格",
     "extend_tools_menu_ssh_client_dock_title": "SSH 用戶端",
     "extend_tools_menu_ai_code_review_dock_title": "AI 程式碼審查",
-    "extend_tools_menu_cot_prompt_editor_dock_title": "CoT 提示詞編輯器",
-    "extend_tools_menu_skill_prompt_editor_dock_title": "Skill 提示詞編輯器",
+    "extend_tools_menu_cot_prompt_editor_dock_title": _COT_PROMPT_EDITOR,
+    "extend_tools_menu_skill_prompt_editor_dock_title": _SKILL_PROMPT_EDITOR,
     "extend_tools_menu_skill_prompt_send_dock_action": "Skill Prompt 傳送停駐窗格",
     "extend_tools_menu_skill_prompt_send_dock_title": "Skill 提示詞傳送 GUI",
     # CoT code-review GUI

pybreeze/pybreeze_ui/connect_gui/ssh/ssh_command_widget.py

@@ -13,6 +13,7 @@
 from je_editor import language_wrapper
 
 from pybreeze.pybreeze_ui.connect_gui.ssh.ssh_host_key_policy import apply_host_key_policy
+from pybreeze.pybreeze_ui.connect_gui.ssh.ssh_key_loader import load_private_key
 from pybreeze.pybreeze_ui.connect_gui.ssh.ssh_login_widget import LoginWidget
 from pybreeze.utils.logging.logger import pybreeze_logger
 
@@ -144,51 +145,53 @@ def connect_ssh(self):
             pybreeze_logger.info("SSH connecting to %s:%s", host, port)
 
             if use_key:
-                if not os.path.exists(key_path):
-                    QMessageBox.warning(
-                        self,
-                        self.word_dict.get("ssh_command_widget_dialog_title_key_error"),
-                        self.word_dict.get("ssh_command_widget_dialog_message_key_file_not_exist"))
+                if not self._authenticate_with_key(host, port, user, key_path, password):
                     return
-                try:
-                    pkey = None
-                    for KeyType in (paramiko.RSAKey, paramiko.Ed25519Key, paramiko.ECDSAKey):
-                        try:
-                            pkey = KeyType.from_private_key_file(key_path, password if password else None)
-                            break
-                        except Exception as error:
-                            pybreeze_logger.debug(f"SSH key type failed: {error}")
-                            continue
-                    if pkey is None:
-                        raise ValueError(
-                            self.word_dict.get(
-                                "ssh_command_widget_error_message_unsupported_private_key"
-                            ))
-                    self.ssh_client.connect(hostname=host, port=port, username=user, pkey=pkey, timeout=10)
-                except Exception as e:
-                    raise RuntimeError(
-                        f"{self.word_dict.get('ssh_command_widget_error_message_key_auth_failed')} {e}") from e
             else:
                 self.ssh_client.connect(
                     hostname=host, port=port, username=user, password=password, timeout=10
                 )
 
-            self.shell_channel = self.ssh_client.invoke_shell(term='xterm', width=120, height=32)
-            self.shell_channel.settimeout(0.0)
-            self.reader_thread = SSHReaderThread(self.shell_channel)
-            self.reader_thread.data_received.connect(self._on_data)
-            self.reader_thread.closed.connect(self._on_closed)
-            self.reader_thread.start()
-            self.login_widget.status_label.setText(
-                self.word_dict.get("ssh_command_widget_log_message_connected"))
-            self.append_text(f"{self.word_dict.get('ssh_command_widget_log_message_connected')}"
-                             f" {host}:{port} as {user}\n")
+            self._start_shell(host, port, user)
         except Exception as e:
             self.login_widget.status_label.setText(
                 self.word_dict.get('ssh_command_widget_status_label_disconnected'))
             self.append_text(f"{self.word_dict.get('ssh_command_widget_log_message_error')} {e}\n")
             self._cleanup()
 
+    def _authenticate_with_key(self, host: str, port: int, user: str, key_path: str, password: str) -> bool:
+        """Perform key-based auth. Returns True on success; False if the key file is missing."""
+        if not os.path.exists(key_path):
+            QMessageBox.warning(
+                self,
+                self.word_dict.get("ssh_command_widget_dialog_title_key_error"),
+                self.word_dict.get("ssh_command_widget_dialog_message_key_file_not_exist"))
+            return False
+        try:
+            pkey = load_private_key(key_path, password, context="SSH")
+            if pkey is None:
+                raise ValueError(
+                    self.word_dict.get(
+                        "ssh_command_widget_error_message_unsupported_private_key"
+                    ))
+            self.ssh_client.connect(hostname=host, port=port, username=user, pkey=pkey, timeout=10)
+        except Exception as e:
+            raise RuntimeError(
+                f"{self.word_dict.get('ssh_command_widget_error_message_key_auth_failed')} {e}") from e
+        return True
+
+    def _start_shell(self, host: str, port: int, user: str) -> None:
+        self.shell_channel = self.ssh_client.invoke_shell(term='xterm', width=120, height=32)
+        self.shell_channel.settimeout(0.0)
+        self.reader_thread = SSHReaderThread(self.shell_channel)
+        self.reader_thread.data_received.connect(self._on_data)
+        self.reader_thread.closed.connect(self._on_closed)
+        self.reader_thread.start()
+        self.login_widget.status_label.setText(
+            self.word_dict.get("ssh_command_widget_log_message_connected"))
+        self.append_text(f"{self.word_dict.get('ssh_command_widget_log_message_connected')}"
+                         f" {host}:{port} as {user}\n")
+
     def _on_data(self, data: bytes):
         try:
             text = data.decode("utf-8", errors="replace")