Merge pull request #90 from Integration-Automation/dev

Dev

Author: JE-Chen

.idea/LoadDensity.iml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "je_load_density_dev"
-version = "0.0.68"
+version = "0.0.71"
 authors = [
     { name = "JE-Chen", email = "[email protected]" },
 ]
@@ -34,6 +34,8 @@ Code = "https://github.com/Intergration-Automation-Testing/LoadDensity"
 file = "README.md"
 content-type = "text/markdown"
 
-
 [tool.setuptools.packages]
 find = { namespaces = false }
+
+[project.optional-dependencies]
+gui = ["PySide6==6.9.1", "qt-material"]

.idea/misc.xml

@@ -0,0 +1,26 @@
+from PySide6.QtCore import QThread
+from je_load_density.wrapper.start_wrapper.start_test import start_test
+
+
+class LoadDensityThread(QThread):
+
+    def __init__(self):
+        super().__init__()
+        self.url = None
+        self.test_time = None
+        self.user_count = None
+        self.spawn_rate = None
+        self.method = None
+
+    def run(self):
+        start_test(
+            {
+                "user": "fast_http_user",
+            },
+            self.user_count, self.spawn_rate, self.test_time,
+            **{
+                "tasks": {
+                    self.method: {"request_url": self.url},
+                }
+            }
+        )

dev.toml

@@ -0,0 +1,20 @@
+english_word_dict = {
+    # Main
+    "application_name": "LoadDensity",
+    # Widget
+    "url": "URL",
+    "test_time": "Test Time: ",
+    "user_count": "User Count: ",
+    "spawn_rate": "Spawn Rate: ",
+    "test_method": "Test Method: ",
+    "start_button": "Start: ",
+    "log": "Log: ",
+    # Test method
+    "get": "GET",
+    "post": "POST",
+    "put": "PUT",
+    "patch": "PATCH",
+    "delete": "DELETE",
+    "head": "HEAD",
+    "options": "OPTIONS",
+}

je_load_density/gui/__init__.py

@@ -0,0 +1,30 @@
+from je_load_density.gui.language_wrapper.english import english_word_dict
+from je_load_density.gui.language_wrapper.traditional_chinese import traditional_chinese_word_dict
+from je_load_density.utils.logging.loggin_instance import load_density_logger
+
+
+
+class LanguageWrapper(object):
+
+    def __init__(
+            self
+    ):
+        load_density_logger.info("Init LanguageWrapper")
+        self.language: str = "English"
+        self.choose_language_dict = {
+            "English": english_word_dict,
+            "Traditional_Chinese": traditional_chinese_word_dict
+        }
+        self.language_word_dict: dict = self.choose_language_dict.get(self.language)
+
+    def reset_language(self, language) -> None:
+        load_density_logger.info(f"LanguageWrapper reset_language language: {language}")
+        if language in [
+            "English",
+            "Traditional_Chinese"
+        ]:
+            self.language = language
+            self.language_word_dict = self.choose_language_dict.get(self.language)
+
+
+language_wrapper = LanguageWrapper()

je_load_density/gui/execute_thread.py

@@ -0,0 +1,20 @@
+traditional_chinese_word_dict = {
+    # Main
+    "application_name": "LoadDensity",
+    # Widget
+    "url": "連結: ",
+    "test_time": "測試時間: ",
+    "user_count": "測試使用者總數: ",
+    "spawn_rate": "產生速率: ",
+    "test_method": "測試 HTTP 方法: ",
+    "start_button": "開始: ",
+    "log": "日誌: ",
+    # Test method
+    "get": "GET",
+    "post": "POST",
+    "put": "PUT",
+    "patch": "PATCH",
+    "delete": "DELETE",
+    "head": "HEAD",
+    "options": "OPTIONS",
+}

je_load_density/gui/language_wrapper/__init__.py

@@ -0,0 +1,81 @@
+import logging
+
+from PySide6.QtCore import QTimer
+from PySide6.QtGui import QIntValidator
+from PySide6.QtWidgets import QWidget, QFormLayout, QLineEdit, QComboBox, QPushButton, QTextEdit, QVBoxLayout, QLabel
+
+from je_load_density.gui.execute_thread import LoadDensityThread
+from je_load_density.gui.language_wrapper.multi_language_wrapper import language_wrapper
+from je_load_density.log_to_ui_filter import InterceptAllFilter, locust_log_queue
+
+
+class LoadDensityWidget(QWidget):
+
+    def __init__(self, parent=None):
+        super().__init__(parent)
+        # from
+        form_layout = QFormLayout()
+        self.url_input = QLineEdit()
+        self.test_time_input = QLineEdit()
+        self.test_time_input.setValidator(QIntValidator())
+        self.user_count_input = QLineEdit()
+        self.user_count_input.setValidator(QIntValidator())
+        self.spawn_rate_input = QLineEdit()
+        self.spawn_rate_input.setValidator(QIntValidator())
+        self.method_combobox = QComboBox()
+        self.method_combobox.addItems([
+            language_wrapper.language_word_dict.get("get"),
+            language_wrapper.language_word_dict.get("post"),
+            language_wrapper.language_word_dict.get("put"),
+            language_wrapper.language_word_dict.get("patch"),
+            language_wrapper.language_word_dict.get("delete"),
+            language_wrapper.language_word_dict.get("head"),
+            language_wrapper.language_word_dict.get("options"),
+        ])
+        form_layout.addRow(language_wrapper.language_word_dict.get("url"), self.url_input)
+        form_layout.addRow(language_wrapper.language_word_dict.get("test_time"), self.test_time_input)
+        form_layout.addRow(language_wrapper.language_word_dict.get("user_count"), self.user_count_input)
+        form_layout.addRow(language_wrapper.language_word_dict.get("spawn_rate"), self.spawn_rate_input)
+        form_layout.addRow(language_wrapper.language_word_dict.get("test_method"), self.method_combobox)
+
+        self.start_button = QPushButton(language_wrapper.language_word_dict.get("start_button"))
+        self.start_button.clicked.connect(self.run_load_density)
+
+        # Log panel
+        self.log_panel = QTextEdit()
+        self.log_panel.setReadOnly(True)
+
+        # Add widget to vertical layout
+        main_layout = QVBoxLayout()
+        main_layout.addLayout(form_layout)
+        main_layout.addWidget(self.start_button)
+        main_layout.addWidget(QLabel(language_wrapper.language_word_dict.get("log")))
+        main_layout.addWidget(self.log_panel)
+
+        # Param
+        self.run_load_density_thread = None
+        self.pull_log_timer = QTimer()
+        self.pull_log_timer.setInterval(20)
+        self.pull_log_timer.timeout.connect(self.add_text_to_log)
+
+        self.setLayout(main_layout)
+
+    def run_load_density(self):
+        self.run_load_density_thread = LoadDensityThread()
+        self.run_load_density_thread.url = self.url_input.text()
+        self.run_load_density_thread.test_time = int(self.test_time_input.text())
+        self.run_load_density_thread.user_count = int(self.user_count_input.text())
+        self.run_load_density_thread.spawn_rate = int(self.spawn_rate_input.text())
+        self.run_load_density_thread.method = self.method_combobox.currentText().lower()
+        log_handler_list = [handler for handler in logging.getLogger("root").handlers if handler.name == "log_reader"]
+        if log_handler_list:
+            log_handler = log_handler_list[0]
+            log_handler.addFilter(InterceptAllFilter())
+        self.run_load_density_thread.start()
+        self.pull_log_timer.stop()
+        self.pull_log_timer.start()
+        self.log_panel.clear()
+
+    def add_text_to_log(self):
+        if not locust_log_queue.empty():
+            self.log_panel.append(locust_log_queue.get_nowait())