Update dev version

Update dev version
- Add GUI

Author: JE-Chen

je_load_density/gui/__init__.py

@@ -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},
+                }
+            }
+        )

je_load_density/gui/execute_thread.py

@@ -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/language_wrapper/__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/language_wrapper/english.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/multi_language_wrapper.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())

je_load_density/gui/language_wrapper/traditional_chinese.py

@@ -0,0 +1,31 @@
+import sys
+
+from PySide6.QtWidgets import QMainWindow, QApplication
+from qt_material import QtStyleTools
+
+from je_load_density.gui.language_wrapper.multi_language_wrapper import language_wrapper
+from je_load_density.gui.main_widget import LoadDensityWidget
+
+
+class LoadDensityUI(QMainWindow, QtStyleTools):
+
+    def __init__(self):
+        super().__init__()
+        self.id = language_wrapper.language_word_dict.get("application_name")
+        if sys.platform in ["win32", "cygwin", "msys"]:
+            from ctypes import windll
+            windll.shell32.SetCurrentProcessExplicitAppUserModelID(self.id)
+        self.setStyleSheet(
+            f"font-size: 12pt;"
+            f"font-family: 'Lato';"
+        )
+        self.apply_stylesheet(self, "dark_amber.xml")
+        self.load_density_widget = LoadDensityWidget()
+        self.setCentralWidget(self.load_density_widget)
+
+
+if __name__ == "__main__":
+    app = QApplication(sys.argv)
+    window = LoadDensityUI()
+    window.show()
+    sys.exit(app.exec())

je_load_density/gui/main_widget.py

@@ -0,0 +1,10 @@
+import logging
+import queue
+
+locust_log_queue = queue.Queue()
+
+class InterceptAllFilter(logging.Filter):
+
+    def filter(self, record):
+        locust_log_queue.put(record.getMessage())
+        return False

je_load_density/gui/main_window.py

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "je_load_density_dev"
-version = "0.0.69"
+version = "0.0.71"
 authors = [
     { name = "JE-Chen", email = "[email protected]" },
 ]
@@ -38,4 +38,4 @@ content-type = "text/markdown"
 find = { namespaces = false }
 
 [project.optional-dependencies]
-gui = ["PySide6==6.9.0"]
+gui = ["PySide6==6.9.1", "qt-material"]