Dev

automation_ide/automation_editor_ui/editor_main/main_ui.py

@@ -1,7 +1,9 @@
 import sys
+from pathlib import Path
 from typing import List, Dict, Type
 
 from PySide6.QtCore import QTimer, QCoreApplication
+from PySide6.QtGui import QIcon
 from PySide6.QtWidgets import QApplication, QWidget
 from je_editor import EditorMain, language_wrapper
 from qt_material import apply_stylesheet
@@ -36,6 +38,9 @@ def __init__(self, debug_mode: bool = False, show_system_tray_ray: bool = False)
         # Tab
         for widget_name, widget in EDITOR_EXTEND_TAB.items():
             self.tab_widget.addTab(widget(), widget_name)
+        # Icon
+        self.icon_path = Path(str(Path.cwd()) + "/je_driver_icon.ico")
+        self.icon = QIcon(str(self.icon_path))
         # Title
         self.setWindowTitle(language_wrapper.language_word_dict.get("automation_editor_application_name"))
         if debug_mode:

automation_ide/automation_editor_ui/extend_multi_language/extend_english.py

@@ -94,5 +94,9 @@ def update_english_word_dict():
             "tools_menu_re_edge_gpt_doc_tab_label": "ReEdgeGPT Doc",
             "tools_menu_re_edge_gpt_github_label": "Open ReEdgeGPT GitHub",
             "tools_menu_re_edge_gpt_github_tab_label": "ReEdgeGPT GitHub",
+            # Test Pioneer Menu
+            "test_pioneer_label": "TestPioneer",
+            "test_pioneer_create_template_label": "Create TestPioneer Yaml template",
+            "test_pioneer_run_yaml": "Execute Test Pioneer Yaml",
         }
     )

automation_ide/automation_editor_ui/extend_multi_language/extend_traditional_chinese.py

@@ -94,5 +94,9 @@ def update_traditional_chinese_word_dict():
             "tools_menu_re_edge_gpt_doc_tab_label": "ReEdgeGPT 文件",
             "tools_menu_re_edge_gpt_github_label": "開啟 ReEdgeGPT GitHub",
             "tools_menu_re_edge_gpt_github_tab_label": "ReEdgeGPT GitHub",
+            # Test Pioneer Menu
+            "test_pioneer_label": "TestPioneer",
+            "test_pioneer_create_template_label": "建立 TestPioneer Yaml 模板",
+            "test_pioneer_run_yaml": "執行 Test Pioneer Yaml",
         }
     )

automation_ide/automation_editor_ui/menu/automation_menu/test_pioneer_menu/build_test_pioneer_menu.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from automation_ide.extend.process_executor.test_pioneer.test_pioneer_process_manager import init_and_start_test_pioneer_process
+
+if TYPE_CHECKING:
+    from automation_ide.automation_editor_ui.editor_main.main_ui import AutomationEditor
+
+from PySide6.QtGui import QAction
+from je_editor import language_wrapper
+from test_pioneer import create_template_dir, execute_yaml
+
+
+def set_test_pioneer_menu(ui_we_want_to_set: AutomationEditor):
+    """
+    Build menu include Test Pioneer feature.
+    :param ui_we_want_to_set: main window to add menu.
+    :return: None
+    """
+    ui_we_want_to_set.test_pioneer_menu = ui_we_want_to_set.automation_menu.addMenu(
+        language_wrapper.language_word_dict.get("test_pioneer_label"))
+    # Create test pioneer template
+    ui_we_want_to_set.create_template_action = QAction(
+        language_wrapper.language_word_dict.get("test_pioneer_create_template_label"))
+    ui_we_want_to_set.create_template_action.triggered.connect(
+        lambda: create_template_dir()
+    )
+    ui_we_want_to_set.test_pioneer_menu.addAction(
+        ui_we_want_to_set.create_template_action
+    )
+    # Run test pioneer yaml
+    ui_we_want_to_set.run_yaml_action = QAction(
+        language_wrapper.language_word_dict.get("test_pioneer_run_yaml"))
+    ui_we_want_to_set.run_yaml_action.triggered.connect(
+        lambda: init_and_start_test_pioneer_process(ui_we_want_to_set)
+    )
+    ui_we_want_to_set.test_pioneer_menu.addAction(
+        ui_we_want_to_set.run_yaml_action
+    )

automation_ide/automation_editor_ui/menu/build_menubar.py

@@ -2,6 +2,9 @@
 
 from typing import TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from automation_ide.automation_editor_ui.editor_main.main_ui import AutomationEditor
+
 from automation_ide.automation_editor_ui.menu.automation_menu.api_testka_menu.build_api_testka_menu import \
     set_apitestka_menu
 from automation_ide.automation_editor_ui.menu.automation_menu.auto_control_menu.build_autocontrol_menu import \
@@ -18,10 +21,8 @@
     build_automation_install_menu
 from automation_ide.automation_editor_ui.menu.install_menu.tools_menu.build_tool_install_menu import \
     build_tool_install_menu
-from automation_ide.automation_editor_ui.menu.tools_menu.bing_gpt_menu.build_bing_gpt_menu import set_bing_gpt_menu
-
-if TYPE_CHECKING:
-    from automation_ide.automation_editor_ui.editor_main.main_ui import AutomationEditor
+from automation_ide.automation_editor_ui.menu.automation_menu.test_pioneer_menu.build_test_pioneer_menu import \
+    set_test_pioneer_menu
 
 from je_editor import language_wrapper
 
@@ -37,6 +38,6 @@ def add_menu_to_menubar(ui_we_want_to_set: AutomationEditor):
     set_load_density_menu(ui_we_want_to_set)
     set_mail_thunder_menu(ui_we_want_to_set)
     set_web_runner_menu(ui_we_want_to_set)
+    set_test_pioneer_menu(ui_we_want_to_set)
     build_automation_install_menu(ui_we_want_to_set)
     build_tool_install_menu(ui_we_want_to_set)
-    set_bing_gpt_menu(ui_we_want_to_set)

automation_ide/automation_editor_ui/syntax/syntax_extend.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from pathlib import Path
 from typing import TYPE_CHECKING
 
 from je_editor import EditorWidget
@@ -27,6 +26,15 @@ def syntax_extend_package(main_window: AutomationEditor) -> None:
                 }
             }
         )
+    syntax_extend_setting_dict.update({".yml": {}})
+    syntax_extend_setting_dict.get(".yml").update(
+        {
+            "test_pioneer": {
+                "words": set(package_keyword_list.get("test_pioneer")),
+                "color": QColor(255, 153, 0)
+            }
+        }
+    )
     widget = main_window.tab_widget.currentWidget()
     if isinstance(widget, EditorWidget):
         widget.code_edit.reset_highlighter()

automation_ide/automation_editor_ui/syntax/syntax_keyword.py

@@ -142,11 +142,22 @@
     "add_command_to_executor", "create_project_dir"
 ]
 
+test_pioneer_keys: list = [
+    "jobs", "pioneer_log",
+    "steps",
+    "name",
+    "run", "with", "run_folder",
+    "wait",
+    "open_url",
+    "open_program", "close_program", "redirect_stdout", "redirect_stderr",
+]
+
 package_keyword_list = {
     "je_auto_control": auto_control_keys,
     "je_load_density": load_density_keys,
     "je_api_testka": api_testka_keys,
     "je_web_runner": web_runner_keys,
     "automation_file": automation_file_keys,
     "mail_thunder": mail_thunder_keys,
+    "test_pioneer": test_pioneer_keys
 }

automation_ide/extend/process_executor/process_executor_utils.py

@@ -8,7 +8,7 @@
 
 from automation_ide.automation_editor_ui.show_code_window.code_window import CodeWindow
 from automation_ide.extend.mail_thunder_extend.mail_thunder_setting import send_after_test
-from automation_ide.extend.process_executor.task_process_manager import TaskProcessManager
+from automation_ide.extend.process_executor.python_task_process_manager import TaskProcessManager
 from automation_ide.utils.exception.exception_tags import wrong_test_data_format_exception_tag
 from automation_ide.utils.exception.exceptions import ITETestExecutorException
 

automation_ide/extend/process_executor/test_pioneer/test_pioneer_process_manager.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import queue
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from queue import Queue
+from typing import Union, TYPE_CHECKING
+
+from PySide6.QtCore import QTimer
+from PySide6.QtWidgets import QWidget
+from je_editor import EditorWidget
+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 automation_ide.automation_editor_ui.show_code_window.code_window import CodeWindow
+
+if TYPE_CHECKING:
+    from automation_ide.automation_editor_ui.editor_main.main_ui import AutomationEditor
+
+
+class TestPioneerProcess(object):
+
+    def __init__(
+            self,
+            main_window: AutomationEditor,
+            exec_str: Union[str, None] = None,
+            program_buffer: int = 1024000,
+            encoding: str = "utf-8",
+    ):
+        self._main_window: AutomationEditor = main_window
+        self._widget: QWidget = main_window.tab_widget.currentWidget()
+        # Code window init
+        self._code_window = CodeWindow()
+        self._main_window.current_run_code_window.append(self._code_window)
+        self._main_window.clear_code_result()
+        self._still_run_program: bool = False
+        self._program_buffer_size = program_buffer
+        self._run_output_queue: Queue = Queue()
+        self._run_error_queue: Queue = Queue()
+        self._read_program_error_output_from_thread: [threading.Thread, None] = None
+        self._read_program_output_from_thread: [threading.Thread, None] = None
+        self._timer: QTimer = QTimer(self._code_window)
+        if isinstance(self._widget, EditorWidget) and exec_str is None:
+            self._test_format_code = self._widget.code_edit.toPlainText()
+        else:
+            self._test_format_code = exec_str
+        if self._main_window.python_compiler is None:
+            # Renew compiler path
+            if sys.platform in ["win32", "cygwin", "msys"]:
+                venv_path = Path(str(Path.cwd()) + "/venv/Scripts")
+            else:
+                venv_path = Path(str(Path.cwd()) + "/venv/bin")
+            self._compiler_path = check_and_choose_venv(venv_path)
+        else:
+            self._compiler_path = main_window.python_compiler
+        if sys.platform in ["win32", "cygwin", "msys"]:
+            args = [
+                self._compiler_path,
+                "-m",
+                "test_pioneer",
+                "-e",
+                self._test_format_code
+            ]
+        else:
+            args = " ".join([f"{self._compiler_path}", "-m test_pioneer", "-e", f"{self._test_format_code}"])
+        self._process: subprocess.Popen = subprocess.Popen(
+            args,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            shell=True,
+            encoding=encoding
+        )
+
+    # Pyside UI update method
+    def pull_text(self):
+        try:
+            self._code_window.code_result.setTextColor(actually_color_dict.get("normal_output_color"))
+            if not self._run_output_queue.empty():
+                output_message = self._run_output_queue.get_nowait()
+                output_message = str(output_message).strip()
+                if output_message:
+                    self._code_window.code_result.append(output_message)
+            self._code_window.code_result.setTextColor(actually_color_dict.get("error_output_color"))
+            if not self._run_error_queue.empty():
+                error_message = self._run_error_queue.get_nowait()
+                error_message = str(error_message).strip()
+                if error_message:
+                    self._code_window.code_result.append(error_message)
+            self._code_window.code_result.setTextColor(actually_color_dict.get("normal_output_color"))
+        except queue.Empty:
+            pass
+        if self._process is not None:
+            if self._process.returncode == 0:
+                if self._timer.isActive():
+                    self._timer.stop()
+                self.exit_program()
+            elif self._process.returncode is not None:
+                if self._timer.isActive():
+                    self._timer.stop()
+                self.exit_program()
+            if self._still_run_program:
+                # poll return code
+                self._process.poll()
+        else:
+            if self._timer.isActive():
+                self._timer.stop()
+
+    # exit program change run flag to false and clean read thread and queue and process
+    def exit_program(self):
+        self._still_run_program = False
+        if self._read_program_output_from_thread is not None:
+            self._read_program_output_from_thread = None
+        if self._read_program_error_output_from_thread is not None:
+            self._read_program_error_output_from_thread = None
+        self.print_and_clear_queue()
+        if self._process is not None:
+            self._process.terminate()
+            self._code_window.code_result.append(f"Task exit with code {self._process.returncode}")
+            self._process = None
+
+    def print_and_clear_queue(self):
+        self._run_output_queue = queue.Queue()
+        self._run_error_queue = queue.Queue()
+
+    def read_program_output_from_process(self):
+        while self._still_run_program:
+            self.process: subprocess.Popen
+            program_output_data = self._process.stdout.read(self._program_buffer_size)
+            if self._process:
+                self._process.stdout.flush()
+            if program_output_data.strip() != "":
+                self._run_output_queue.put(program_output_data)
+
+    def read_program_error_output_from_process(self):
+        while self._still_run_program:
+            program_error_output_data = self._process.stderr.read(self._program_buffer_size)
+            if self._process:
+                self._process.stderr.flush()
+            if program_error_output_data.strip() != "":
+                self._run_error_queue.put(program_error_output_data)
+
+    def start_test_pioneer_process(self):
+        self._still_run_program = True
+        # program output message queue thread
+        self._read_program_output_from_thread = threading.Thread(
+            target=self.read_program_output_from_process,
+            daemon=True
+        )
+        self._read_program_output_from_thread.start()
+        # program error message queue thread
+        self._read_program_error_output_from_thread = threading.Thread(
+            target=self.read_program_error_output_from_process,
+            daemon=True
+        )
+        self._read_program_error_output_from_thread.start()
+        # start Pyside update
+        # start timer
+        self._code_window.setWindowTitle("Test Pioneer")
+        self._code_window.show()
+        self._timer = QTimer()
+        self._timer.setInterval(100)
+        self._timer.timeout.connect(self.pull_text)
+        self._timer.start()
+
+
+def init_and_start_test_pioneer_process(ui_we_want_to_set: AutomationEditor):
+    test_pioneer_process_manager = TestPioneerProcess(main_window=ui_we_want_to_set)
+    test_pioneer_process_manager.start_test_pioneer_process()

dev_requirements.txt

@@ -8,4 +8,6 @@ je-mail-thunder
 automation_ide_dev
 sphinx
 sphinx-rtd-theme
-PySide6==6.7.3
+PySide6==6.8.2.1
+auto-py-to-exe
+test_pioneer