From a353511f6459998905cdc3f933853197dfe16160 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois?= Date: Sat, 18 Apr 2026 13:32:45 +0200 Subject: [PATCH] Split MainWindow into TestRunner and TestFileManager coordinators Extract execution lifecycle (TestRunner) and file/process management (TestFileManager) from MainWindow, reducing it from ~1170 to ~700 lines. Validated against full test suite with no new regressions. Co-Authored-By: Claude Sonnet 4.6 --- src/testium/main_win/test_file_manager.py | 171 ++ src/testium/main_win/test_runner.py | 237 +++ src/testium/main_win/testium_win.py | 1872 ++++++++------------- 3 files changed, 1109 insertions(+), 1171 deletions(-) create mode 100644 src/testium/main_win/test_file_manager.py create mode 100644 src/testium/main_win/test_runner.py diff --git a/src/testium/main_win/test_file_manager.py b/src/testium/main_win/test_file_manager.py new file mode 100644 index 0000000..4fd290d --- /dev/null +++ b/src/testium/main_win/test_file_manager.py @@ -0,0 +1,171 @@ +import os +import sys +import traceback +from queue import Empty + +from PySide6.QtWidgets import QApplication, QFileDialog + +from interpreter.process import TestProcess +from interpreter.utils.test_ctrl import TestSetController +from main_win.test_controller_service import TestControllerService +import interpreter.utils.settings as prefs +from lib.tum_except import ETUMFileError, ETUMRuntimeError + + +class TestFileManager: + """Manages test file loading, process lifecycle, and recent files.""" + + def __init__(self, win) -> None: + self._win = win + + # --- Process lifecycle --- + + def clear_process(self): + w = self._win + if ( + w.test_proc is not None + and w.test_proc.is_alive() + and w.test_service is not None + ): + w.test_service.stop() + w.test_service.close() + w.test_proc.join() + del w.test_proc + w.test_proc = None + del w.test_service + w.test_service = None + del w.ts_controller + w.ts_controller = None + + def reload(self, file_name: str): + w = self._win + w.disconnect_signals() + self.clear_process() + self.load(file_name) + w.reconnect_signals() + + def load(self, file_name: str) -> bool: + """Load a test file. Returns True on success, False otherwise.""" + w = self._win + try: + if not file_name: + raise ETUMFileError("No file to load") + + file_name = os.path.abspath(file_name) + initial_dir = os.path.dirname(file_name) + + if not os.path.isdir(initial_dir): + raise ETUMFileError("Could not find %s directory" % initial_dir) + if not os.path.isfile(file_name): + raise ETUMFileError("Could not find %s file" % file_name) + + w.testFile = None + w.ts_controller = TestSetController() + w.test_service = TestControllerService(w.ts_controller) + w.test_proc = TestProcess( + file_name, + w.status_queue, + w.ts_controller, + w.config_files, + w.defines, + self._defaults_for_process(), + ) + w.test_proc.start() + while w.test_proc.is_alive(): + try: + if w.test_service.loaded(timeout=1.0): + break + except Empty: + w.test_service.clear() + + if not w.test_proc.is_alive(): + del w.test_proc + w.test_proc = None + del w.test_service + w.test_service = None + del w.ts_controller + w.ts_controller = None + raise ETUMRuntimeError( + "Test could not be loaded (test process crashed for any reason)" + ) + + test_data = w.test_service.tree() + w.treeTests.clear() + w.treeTests.loadTestRecursively(w.treeTests.invisibleRootItem(), test_data) + w.treeTests.setFoldDefault() + w.treeTests.updateTreeSkipState(w.test_service) + + w.checkSelect.setChecked(True) + w.testFile = file_name + test_dir = os.path.dirname(w.testFile) + + sys.path.append(test_dir) + w.statusBar().showMessage("Test file loaded", 10000) + w.textLog.set_test_dir(test_dir) + self.add_file_to_recent(file_name) + w.setWindowTitle(w.mainWindowTitle + " - " + w.testFile) + w.actionStart_test.setEnabled(True) + w.actionRefresh_test.setEnabled(True) + w.show_checkboxes() + return True + except: + w.statusBar().showMessage("No test file could be loaded", 10000) + w.treeTests.clear() + print(traceback.format_exc()) + return False + + def _defaults_for_process(self) -> dict: + d = {} + pp = prefs.settings.python_bin + if pp != "": + d["python_bin"] = pp + pp = prefs.settings.lua_bin + if pp != "": + d["lua_bin"] = pp + return d + + # --- Recent files --- + + def add_file_to_recent(self, filename: str): + files = prefs.settings.recent_files + try: + files.remove(filename) + except ValueError: + pass + files.insert(0, filename) + del files[self._win.MaxRecentFiles:] + prefs.settings.recent_files = files + for widget in QApplication.topLevelWidgets(): + from main_win.testium_win import MainWindow + if isinstance(widget, MainWindow): + widget.file_manager.update_recent_file_actions() + + def update_recent_file_actions(self): + w = self._win + files = prefs.settings.recent_files + numRecentFiles = min(len(files), w.MaxRecentFiles) + for i in range(numRecentFiles): + text = "&%d %s" % (i + 1, w._stripped_name(files[i])) + w.recentFileActs[i].setText(text) + w.recentFileActs[i].setData(files[i]) + w.recentFileActs[i].setVisible(True) + for j in range(numRecentFiles, w.MaxRecentFiles): + w.recentFileActs[j].setVisible(False) + w.separatorAct.setVisible(numRecentFiles > 0) + + def on_open_recent_file(self): + w = self._win + action = w.sender() + if action: + self.reload(action.data()) + + def on_open_test(self): + w = self._win + d = "" + if w.testFile is not None: + d = os.path.dirname(w.testFile) + file_name, _ = QFileDialog.getOpenFileName( + w, "Open the test file", d, "testium file (*.tum);;All Files (*)" + ) + if file_name: + self.reload(file_name) diff --git a/src/testium/main_win/test_runner.py b/src/testium/main_win/test_runner.py new file mode 100644 index 0000000..b42ad00 --- /dev/null +++ b/src/testium/main_win/test_runner.py @@ -0,0 +1,237 @@ +import os +import traceback +from tempfile import NamedTemporaryFile + +from PySide6 import QtGui +from PySide6.QtCore import QDateTime +from PySide6.QtGui import QIcon, QPixmap + +from interpreter.utils.icons import icon_prefix +import interpreter.utils.settings as prefs + + +class TestRunner: + """Manages the test execution lifecycle: start/pause/stop, timers, log file, UI adaptation.""" + + def __init__(self, win) -> None: + self._win = win + self.logFileHandler = None + + # --- Timer helpers --- + + def start_pause_timer(self): + w = self._win + w.timerPause.setSingleShot(False) + w.timerPause.setInterval(500) + w.timerPause.start() + w.timerPause.state = False + + # --- Execution control --- + + def on_start_test(self): + w = self._win + + if w._test_started: + if not w._test_paused: + w.test_service.pause() + self.start_pause_timer() + else: + w.test_service.cont() + w.timerPause.stop() + w.timerPause.state = False + self.on_timer_pause() + w._test_paused = not w._test_paused + return + + w.start_time = QDateTime.currentDateTime() + + # Log file setup + log_file = w.editLogFilePath.text() + if w.buttLogFileSaved.isChecked() and (log_file != ""): + try: + if not os.path.isabs(log_file): + default_path = prefs.settings.log_path + default_path = w.test_service.process_param(default_path) + log_file = os.path.join(default_path, log_file) + if not os.path.exists(os.path.dirname(log_file)): + os.makedirs(os.path.dirname(log_file)) + if os.path.isfile(log_file): + i = 0 + fname = log_file + while os.path.isfile(fname): + i += 1 + fname = log_file + "-" + str(i) + ".saved" + os.rename(log_file, fname) + self.logFileHandler = open(log_file, "w") + w.out_log.set(self.logFileHandler) + w.logFileName = log_file + except: + self.logFileHandler = NamedTemporaryFile(mode="w", suffix=".log", delete=False) + w.out_log.set(self.logFileHandler) + w.logFileName = self.logFileHandler.name + else: + self.logFileHandler = NamedTemporaryFile(mode="w", suffix=".log", delete=False) + w.out_log.set(self.logFileHandler) + w.logFileName = self.logFileHandler.name + + # Report setup and execution + rep_file = w.test_service.process_param(w.reportFileName) + w.test_service.set_report(rep_file, w.report_type, w.report_pattern) + self.adapt_interface_during_test() + w.treeTests.clearAllStatus() + try: + w.textLog.clear() + w.textLog.appendPlainText("Test is started\n") + w.timer.setSingleShot(False) + w.timer.setInterval(100) + w.timer.start() + w.test_service.set_test_outputs([w.logFileName]) + w.test_service.execute() + except: + print(traceback.format_exc()) + self.restore_interface_after_test() + + def on_stop_test(self): + self._win.test_service.stop() + + def on_run_finished(self): + w = self._win + w.timer.setSingleShot(True) + w.timer.setInterval(1000) + txt = w.stream.read() + w.textLog.appendPlainText(txt) + self.restore_interface_after_test() + + if self.logFileHandler is not None: + w.out_log.reset() + self.logFileHandler.write(txt + "\n") + self.logFileHandler.close() + self.logFileHandler = None + + w.textLog.appendPlainText("Test is finished") + if w.runandclose: + w.on_actionExit_triggered() + + def on_breakpoint(self): + w = self._win + w._test_paused = True + self.start_pause_timer() + + # --- Timer slots --- + + def on_timer_event(self): + w = self._win + text_to_append = [] + while not w.threads_queue.empty(): + text_to_append.append(w.threads_queue.get()) + if text_to_append: + for t in text_to_append: + w.textLog.appendPlainText(t) + if self.logFileHandler is not None: + self.logFileHandler.write(t + "\n") + self.logFileHandler.flush() + + def on_timer_blink(self): + w = self._win + if w.buttBlink.current_color != "gray": + self.set_blink_gray() + elif w.treeTests.getGlobalSuccess(): + self.set_blink_green() + else: + self.set_blink_red() + + def on_timer_pause(self): + w = self._win + if w._test_paused: + icon = QtGui.QIcon() + if w.timerPause.state: + icon.addPixmap(QtGui.QPixmap(icon_prefix() + "/pause2.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off) + else: + icon.addPixmap(QtGui.QPixmap(icon_prefix() + "/pause.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off) + w.timerPause.state = not w.timerPause.state + w.actionStart_test.setIcon(icon) + + def on_timer_count(self): + w = self._win + secfromstart = w.start_time.secsTo(QDateTime.currentDateTime()) + w.label_runtime.setText( + "%02d:%02d:%02d" % (secfromstart / 3600, (secfromstart / 60) % 60, secfromstart % 60) + ) + + # --- Interface adaptation --- + + def adapt_interface_during_test(self): + w = self._win + try: + w.disconnect_signals() + w.actionOpenTest.setDisabled(True) + w.actionExit.setDisabled(True) + icon = QtGui.QIcon() + icon.addPixmap(QtGui.QPixmap(icon_prefix() + "/pause.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off) + w.actionStart_test.setIcon(icon) + w.actionStart_test.setText("Pause test") + w.actionPreferences.setDisabled(True) + w.actionRefresh_test.setDisabled(True) + w.actionShow_Results.setDisabled(True) + w.actionSave_report.setDisabled(True) + w.logSettingsBox.setDisabled(True) + w.actionStop_test.setEnabled(True) + if prefs.settings.show_checkboxes: + w._checklist = w.treeTests.getCheckList() + w.treeTests.removeCheckBoxes() + w.checkSelect.setDisabled(True) + w.checkFold.setDisabled(True) + w.timerBlink.setSingleShot(False) + w.timerBlink.setInterval(1000) + w.timerBlink.start() + self.set_blink_green() + w.treeTests.clearGlobalSuccess() + finally: + w._test_started = True + + def restore_interface_after_test(self): + w = self._win + try: + w.timerPause.stop() + w.timerBlink.stop() + w.actionOpenTest.setEnabled(True) + w.actionExit.setEnabled(True) + icon = QtGui.QIcon() + icon.addPixmap(QtGui.QPixmap(icon_prefix() + "/start.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off) + w.actionStart_test.setIcon(icon) + w.actionStart_test.setText("Start test") + w.actionPreferences.setEnabled(True) + w.actionRefresh_test.setEnabled(True) + w.actionStop_test.setDisabled(True) + w.actionShow_Results.setEnabled(True) + w.actionSave_report.setEnabled(True) + w.logSettingsBox.setEnabled(True) + if prefs.settings.show_checkboxes: + w.checkSelect.setEnabled(True) + w.treeTests.showCheckBoxes(w._checklist, w.test_service) + w.checkFold.setEnabled(True) + w.treeTests.setChildrenEnabled() + w.reconnect_signals() + if w.treeTests.getGlobalSuccess(): + self.set_blink_green() + else: + self.set_blink_red() + finally: + w._test_started = False + + # --- Blink indicator --- + + def set_blink_green(self): + w = self._win + w.buttBlink.setIcon(w.iconBlinkGreen) + w.buttBlink.current_color = "green" + + def set_blink_red(self): + w = self._win + w.buttBlink.setIcon(w.iconBlinkRed) + w.buttBlink.current_color = "red" + + def set_blink_gray(self): + w = self._win + w.buttBlink.setIcon(w.iconBlinkGray) + w.buttBlink.current_color = "gray" diff --git a/src/testium/main_win/testium_win.py b/src/testium/main_win/testium_win.py index 82a4041..792576c 100755 --- a/src/testium/main_win/testium_win.py +++ b/src/testium/main_win/testium_win.py @@ -1,1171 +1,701 @@ -import sys -import os -import subprocess -import traceback -import webbrowser -from time import sleep -from tempfile import NamedTemporaryFile -from multiprocessing import Queue -from queue import Empty -from threading import Thread -import shutil -import ast - -# Qt -from PySide6 import QtGui, QtWidgets -from PySide6.QtGui import QAction, QShortcut, QIcon, QPixmap, QTextCursor, QDesktopServices, QTextCursor -from PySide6.QtCore import Slot, QUrl, Qt, QTimer, QDateTime - -from PySide6.QtWidgets import ( - QApplication, - QMainWindow, - QDialog, - QFileDialog, - QSizePolicy, -) - -ourPath = os.path.dirname(__file__) -sys.path.append(os.path.join(ourPath, "resources")) - -# user interfaces -from main_win.testium_core_win import Ui_MainWindow -from main_win.text_log import QTextLog -from main_win.about_win.about_win import Ui_About -from main_win.preference_win.preference_win import PrefWindow -from main_win.f1_win.d_f1_win import DialogF1 -from main_win.test_tree import QTestTree - -from main_win.test_run.thread_output import ThreadTestOutput -from lib.string_queue import StringQueue -from interpreter.process import TestProcess -from interpreter.utils.test_ctrl import TestSetController -from main_win.test_controller_service import TestControllerService -from interpreter.utils.icons import icon_prefix - -from main_win.test_run.outlog import OutLog -from main_win.test_run.test_run import ThreadTestStatus -import interpreter.utils.settings as prefs -from lib.stdout_redirect import stdio_redir -import libs.testium as tm -from interpreter.utils.version import get_testium_version -from interpreter.utils.test_init import ( - env_init, - locate_report_file, -) -from lib.tum_except import ETUMFileError, ETUMRuntimeError - - -class MainWindow(QMainWindow, Ui_MainWindow): - MaxRecentFiles = 5 - - def __init__( - self, - test_file=None, - config_files="", - runandclose=False, - log_file="", - defines={}, - report="", - report_type="", - report_pattern=[], - debug=False, - ): - super().__init__() - self.setupUi(self) - self.textLog = self.create_text_log(self.frame1) - self.verticalLayout_2.addWidget(self.textLog) - - icon2 = QtGui.QIcon() - icon2.addPixmap( - QtGui.QPixmap(icon_prefix() + "/edit-clear.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.buttClearLog.setIcon(icon2) - icon3 = QtGui.QIcon() - icon3.addPixmap( - QtGui.QPixmap(icon_prefix() + "/go-bottom.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.buttGoBottom.setIcon(icon3) - icon4 = QtGui.QIcon() - icon4.addPixmap( - QtGui.QPixmap(icon_prefix() + "/document-open.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionOpenTest.setIcon(icon4) - icon5 = QtGui.QIcon() - icon5.addPixmap( - QtGui.QPixmap(icon_prefix() + "/document-save.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionSave_report.setIcon(icon5) - icon6 = QtGui.QIcon() - icon6.addPixmap( - QtGui.QPixmap(icon_prefix() + "/start.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionStart_test.setIcon(icon6) - icon7 = QtGui.QIcon() - icon7.addPixmap( - QtGui.QPixmap(icon_prefix() + "/stop.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionStop_test.setIcon(icon7) - icon8 = QtGui.QIcon() - icon8.addPixmap( - QtGui.QPixmap(icon_prefix() + "/about.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionAbout_testium.setIcon(icon8) - icon9 = QtGui.QIcon() - icon9.addPixmap( - QtGui.QPixmap(icon_prefix() + "/exit.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionExit.setIcon(icon9) - icon10 = QtGui.QIcon() - icon10.addPixmap( - QtGui.QPixmap(icon_prefix() + "/view-refresh.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionRefresh_test.setIcon(icon10) - icon11 = QtGui.QIcon() - icon11.addPixmap( - QtGui.QPixmap(icon_prefix() + "/results.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionShow_Results.setIcon(icon11) - icon12 = QtGui.QIcon() - icon12.addPixmap( - QtGui.QPixmap(icon_prefix() + "/help.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionHelp.setIcon(icon12) - icon13 = QtGui.QIcon() - icon13.addPixmap( - QtGui.QPixmap(icon_prefix() + "/settings.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionPreferences.setIcon(icon13) - - icon14 = QtGui.QIcon() - icon14.addPixmap( - QtGui.QPixmap(icon_prefix() + "/info.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionTestInformation.setIcon(icon14) - - self.runandclose = runandclose - # Var init - self.mainWindowTitle = self.windowTitle() - self.logFileHandler = None - self.defines = defines - self.logFileName = log_file - self.reportFileName = report - self.report_type = report_type - self.report_pattern = report_pattern - self.config_files = config_files - self.recentFileActs = [] - self.debug = debug - self.test_proc = None - self.ts_controller = None - self.test_service = None - self.threadTestStatus = None - self._test_started = False - self._test_paused = False - self._signals_connected = False - - self.timer = QTimer() - self.timer.setSingleShot(False) - self.timer.stop() - self.timer.setInterval(100) - - self.timerBlink = QTimer() - self.timerBlink.setSingleShot(False) - self.timerBlink.stop() - self.timerBlink.setInterval(1000) - self.timerPause = QTimer() - self.timerPause.setSingleShot(False) - self.timerPause.stop() - self.timerPause.setInterval(500) - self.timerPause.state = False - self.iconBlinkGreen = QIcon() - self.iconBlinkGreen.addPixmap(QPixmap(icon_prefix() + "/green.png")) - self.iconBlinkRed = QIcon() - self.iconBlinkRed.addPixmap(QPixmap(icon_prefix() + "/red.png")) - self.iconBlinkGray = QIcon() - self.iconBlinkGray.addPixmap(QPixmap(icon_prefix() + "/gray.png")) - self.setBlinkGreen() - - self.threads_queue = Queue() - self.status_queue = Queue() - - env_init() - - # Persistence - self.pref_win = PrefWindow(self) - - lastLog = prefs.settings.log_file - if self.logFileName == "": - self.editLogFilePath.setText(lastLog) - self.logFileName = lastLog - if prefs.settings.log_file_saved: - self.buttLogFileSaved.setChecked(True) - else: - if not os.path.isabs(self.logFileName): - self.logFileName = os.path.join(os.getcwd(), self.logFileName) - self.buttLogFileSaved.setChecked(True) - self.editLogFilePath.setText(self.logFileName) - - geo_settings = prefs.settings.value( - prefs.SettingsItem("geometry", bytearray), bytearray() - ) - if geo_settings: - self.restoreGeometry(geo_settings) - - state_settings = prefs.settings.value( - prefs.SettingsItem("state", bytearray), bytearray() - ) - if state_settings: - self.restoreState(state_settings) - - # disable the action buttons - self.actionStart_test.setDisabled(True) - self.actionShow_Results.setDisabled(True) - self.actionSave_report.setDisabled(True) - - # Tree Test - self.create_tree() - - # Shortcuts - self.shorcut_stop = QShortcut( - Qt.Key_Space, - self.treeTests, - context=Qt.WidgetShortcut, - activated=self.on_spacePressed, - ) - self.shorcut_f1 = QShortcut( - Qt.Key_F1, - self.treeTests, - context=Qt.WidgetShortcut, - activated=self.on_F1Pressed, - ) - - # Main Window items modifications - self.actionRefresh_test.setDisabled(True) - - # Connection of the handlers - self.buttLogFilePath.pressed.connect(self.on_buttLogFilePath_clicked) - self.buttClearLog.pressed.connect(self.on_buttClearLog_clicked) - self.buttGoBottom.pressed.connect(self.on_buttGoBottom_clicked) - self.editLogFilePath.editingFinished.connect(self.on_configLog_changed) - self.buttLogFileSaved.toggled.connect(self.on_configLogSaved_changed) - self.buttLogFileNone.toggled.connect(self.on_configLogNone_changed) - self.timer.timeout.connect(self.on_timerEvent) - self.timerBlink.timeout.connect(self.on_timerBlinkEvent) - self.timerBlink.timeout.connect(self.on_timerCount) - self.timerPause.timeout.connect(self.on_timerPause) - self.treeTests.itemSelectionChanged.connect(self.on_testSelectionChanged) - if prefs.settings.dbl_click_enabled: - self.treeTests.setExpandsOnDoubleClick(False) - self.treeTests.itemDoubleClicked.connect(self.on_testItemDblClicked) - else: - self.treeTests.setExpandsOnDoubleClick(True) - QApplication.instance().lastWindowClosed.connect(self.on_exiting) - - self.prefs_apply_font() - self.prefs_apply_font_size() - - # Recent files - for i in range(MainWindow.MaxRecentFiles): - self.recentFileActs.append( - QAction(self, visible=False, triggered=self.on_openRecentFile) - ) - self.separatorAct = self.menuFile.addSeparator() - for i in range(MainWindow.MaxRecentFiles): - self.menuFile.addAction(self.recentFileActs[i]) - self.updateRecentFileActions() - - # A propos - self.d_about_win = QDialog() - self.about_win = Ui_About() - - self.about_win.setupUi(self.d_about_win) - self.about_win.labelVersion.setText("testium - " + get_testium_version()) - self.about_win.labelCesUnitVersion.setText("") - self.d_about_win.setModal(True) - - # F1 window - self.d_f1_win = DialogF1(self) - - self.stream = StringQueue() # stream used to log output - stdio_redir.redirect(self.stream) - self.threadOutput = ThreadTestOutput(self.stream, self.threads_queue) - self.threadOutput.start() - - self.out_log = OutLog() - self.out_log.logToBeAppended.connect(self.on_logToBeAppended) - self.redirectStdToTextLog(self.out_log) - self.testFile = test_file - - self.threadTestStatus = ThreadTestStatus(self.status_queue, debug=self.debug) - self.threadTestStatus.start() - - self.update_from_prefs() - - # report file name treatment - self.reportFileName = locate_report_file(self.reportFileName) - - # open the last opened file if it exists. - - last_files = prefs.settings.recent_files - ret = False - if test_file != "": - if not os.path.isabs(test_file): - test_file = os.path.join(os.getcwd(), test_file) - if os.path.isfile(test_file): - ret = self.loadTestSetFile(test_file) - elif (len(last_files) > 0) and os.path.isfile(last_files[0]): - ret = self.loadTestSetFile(last_files[0]) - - # In case of successfull loading of a file, we need to update the fold and checked state - if ret: - self.file_loaded_at_startup() - - # connect the test status - self.threadTestStatus.testSetIsFinished.connect(self.on_runFinished) - self.threadTestStatus.statusToBeUpdated.connect(self.treeTests.updateStatus) - self.reconnect_signals() - - if runandclose: - self.on_actionStart_test_triggered() - - def create_text_log(self, parent): - textLog = QTextLog(parent) - return textLog - - def create_tree(self): - self.treeTests = QTestTree(self.widget) - self.treeTests.setEnabled(True) - sizePolicy = QSizePolicy(QSizePolicy.Minimum, QSizePolicy.Expanding) - sizePolicy.setHorizontalStretch(0) - sizePolicy.setVerticalStretch(0) - sizePolicy.setHeightForWidth(self.treeTests.sizePolicy().hasHeightForWidth()) - self.treeTests.setSizePolicy(sizePolicy) - self.treeTests.breakpoint.connect(self.on_breakpoint) - self.verticalLayout.addWidget(self.treeTests) - - def remove_tree(self): - self.verticalLayout.removeWidget(self.treeTests) - del self.treeTests - self.treeTests = None - - def file_loaded_at_startup(self): - modeSlider_value = prefs.settings.show_checkboxes - - # Apply production/Lab state - if modeSlider_value: - # restore check boxes state if in lab mode - checkList = prefs.settings.value(prefs.SettingsItem("checkList", list), []) - if checkList is not None: - if len(checkList) == self.treeTests.getItemCount(): - self.treeTests.restoreCheckList(checkList, self.test_service) - - else: - tm.print_info( - "The number of tests has changed. Test box states are not restored." - ) - - # Apply treeview visibility - foldList = prefs.settings.value(prefs.SettingsItem("foldList", list), []) - if foldList: - if len(foldList) == self.treeTests.getItemCount(): - self.checkFold.setCheckState(Qt.PartiallyChecked) - self.treeTests.restoreFoldList(foldList) - - def disconnect_signals(self): - if self._signals_connected: - # disconnect the GUI - self.checkSelect.stateChanged.disconnect() - self.treeTests.itemChanged.disconnect() - self.checkFold.stateChanged.disconnect() - self.treeTests.itemCollapsed.disconnect() - self.treeTests.itemExpanded.disconnect() - self._signals_connected = False - - def reconnect_signals(self): - if not self._signals_connected: - # reconnect the GUI - self.checkSelect.stateChanged.connect(self.on_selectDeselectAll) - self.treeTests.itemChanged.connect(self.on_testChecked) - self.checkFold.stateChanged.connect(self.on_checkFoldChanged) - self.treeTests.itemCollapsed.connect(self.on_itemFoldChanged) - self.treeTests.itemExpanded.connect(self.on_itemFoldChanged) - self._signals_connected = True - - def prefs_apply_font(self): - f = self.textLog.font() - f.fromString(prefs.settings.log_font) - self.textLog.setFont(f) - - def prefs_apply_font_size(self): - f = self.textLog.font() - f.setPointSize(prefs.settings.log_font_size) - self.textLog.setFont(f) - - def reload_test_set_file(self, file_name: str): - self.disconnect_signals() - self.clear_process() - self.loadTestSetFile(file_name) - self.reconnect_signals() - - def clear_process(self): - if ( - self.test_proc is not None - and self.test_proc.is_alive() - and (self.ts_controller is not None) - ): - self.test_service.stop() - self.test_service.close() - self.test_proc.join() - del self.test_proc - self.test_proc = None - del self.test_service - self.test_service = None - del self.ts_controller - self.ts_controller = None - - @Slot() - def on_actionOpenTest_triggered(self): - d = "" - if self.testFile is not None: - d = os.path.dirname(self.testFile) - file_name, _ = QFileDialog.getOpenFileName( - self, "Open the test file", d, "testium file (*.tum);;All Files (*)" - ) - if file_name: - self.reload_test_set_file(file_name) - - def startPauseTimer(self): - self.timerPause.setSingleShot(False) - self.timerPause.setInterval(500) - self.timerPause.start() - self.timerPause.state = False - - @Slot() - def on_actionStart_test_triggered(self): - # Test to be paused - if self._test_started: - if not self._test_paused: - self.test_service.pause() - self.startPauseTimer() - else: - - # Test to be continued - self.test_service.cont() - self.timerPause.stop() - self.timerPause.state = False - self.on_timerPause() - - self._test_paused = not self._test_paused - return - - # Test to be started - self.start_time = QDateTime.currentDateTime() - - # log file definition - log_file = self.editLogFilePath.text() - if self.buttLogFileSaved.isChecked() and (log_file != ""): - try: - if not os.path.isabs(log_file): - default_path = prefs.settings.log_path - default_path = self.test_service.process_param(default_path) - log_file = os.path.join(default_path, log_file) - # if the directory does not exist - if not os.path.exists(os.path.dirname(log_file)): - os.makedirs(os.path.dirname(log_file)) - # If the file exists - if os.path.isfile(log_file): - i = 0 - fname = log_file - while os.path.isfile(fname): - i += 1 - fname = log_file + "-" + str(i) + ".saved" - - os.rename(log_file, fname) - self.logFileHandler = open(log_file, "w") - self.out_log.set(self.logFileHandler) - self.logFileName = log_file - except: - self.logFileHandler = NamedTemporaryFile( - mode="w", suffix=".log", delete=False - ) - self.out_log.set(self.logFileHandler) - self.logFileName = self.logFileHandler.name - else: - self.logFileHandler = NamedTemporaryFile( - mode="w", suffix=".log", delete=False - ) - self.out_log.set(self.logFileHandler) - self.logFileName = self.logFileHandler.name - - # Report file definition - rep_file = self.test_service.process_param(self.reportFileName) - self.test_service.set_report(rep_file, self.report_type, self.report_pattern) - self.adaptInterfaceDuringTest() - self.treeTests.clearAllStatus() - try: - self.textLog.clear() - self.textLog.appendPlainText("Test is started\n") - self.timer.setSingleShot(False) - self.timer.setInterval(100) - self.timer.start() - # Add the log file to the std test_outputs - self.test_service.set_test_outputs([self.logFileName]) - # Launch the test - self.test_service.execute() - except: - print(traceback.format_exc()) - self.restoreInterfaceAfterTest() - - def on_runFinished(self): - self.timer.setSingleShot(True) - self.timer.setInterval(1000) - txt = self.stream.read() - self.textLog.appendPlainText(txt) - self.restoreInterfaceAfterTest() - - if self.logFileHandler is not None: - self.out_log.reset() - self.logFileHandler.write(txt + "\n") - self.logFileHandler.close() - - self.logFileHandler = None - - self.textLog.appendPlainText("Test is finished") - if self.runandclose: - self.on_actionExit_triggered() - - @Slot() - def on_actionStop_test_triggered(self): - self.test_service.stop() - - def save_settings(self): - prefs.settings.set_value( - prefs.SettingsItem("geometry", bytearray), bytearray(self.saveGeometry()) - ) - prefs.settings.set_value( - prefs.SettingsItem("state", bytearray), bytearray(self.saveState()) - ) - prefs.settings.set_value( - prefs.SettingsItem("checkList", list), self.treeTests.getCheckList() - ) - prefs.settings.set_value( - prefs.SettingsItem("foldList", list), self.treeTests.getFoldList() - ) - self.treeTests.saveSizes() - prefs.settings.sync() - - @Slot() - def on_actionExit_triggered(self): - self.close() - - def on_exiting(self): - if not self._test_started: - self.save_settings() - self.clear_process() - self.threadTestStatus.stop() - self.threadOutput.stop() - self.threadOutput.wait() - self.threadTestStatus.wait() - - @Slot() - def on_actionAbout_testium_triggered(self): - self.d_about_win.setVisible(True) - - @Slot() - def on_actionPreferences_triggered(self): - result = self.pref_win.exec() - if result == QDialog.Accepted: - self.update_from_prefs() - if self.pref_win.isChanged(prefs.settings.SettingsShowCheckboxes): - self.show_checkboxes() - if self.pref_win.isChanged(prefs.settings.SettingsDblClickEnabled): - if prefs.settings.dbl_click_enabled: - self.treeTests.itemDoubleClicked.connect(self.on_testItemDblClicked) - else: - self.treeTests.itemDoubleClicked.disconnect() - if self.pref_win.isChanged(prefs.settings.SettingsLogFont): - self.prefs_apply_font() - if self.pref_win.isChanged(prefs.settings.SettingsLogFontSize): - self.prefs_apply_font_size() - - @Slot() - def on_actionRefresh_test_triggered(self): - self.on_exiting() - args = [] - if not hasattr(sys, "frozen"): - args += [sys.executable] - args += [sys.argv[0]] - - if len(self.defines) > 0: - for k, v in self.defines.items(): - try: - val = ast.literal_eval(v) - except: - val = v - - args += ["-d", f"{k}={val}"] - - if (self.testFile is not None) and (isinstance(self.testFile, str)): - args += [self.testFile] - - os.execv(sys.executable, args) - - @Slot() - def on_actionSave_report_triggered(self): - - if self.testFile: - initialPath = os.path.dirname(self.testFile) - else: - initialPath = None - - fileName, _ = QFileDialog.getSaveFileName( - self, "Path to Log file", initialPath, "Log Files (*.log);;All Files (*)" - ) - if fileName: - shutil.copy(self.logFileName, fileName) - - @Slot() - def on_actionShow_Results_triggered(self): - s = sys.platform - self.statusBar().showMessage( - "Opening the logfile (" + s + "): " + self.logFileName, 100000 - ) - QDesktopServices.openUrl(QUrl.fromLocalFile(self.logFileName)) - - @Slot() - def on_actionHelp_triggered(self): - self.webbrowser_open() - - def webbrowser_open(self): - def open_browser_thread(): - webbrowser.open( - "https://git.beafrancois.fr/v-and-v/testium/src/branch/main/doc/manual/testium_manual.pdf", - new=2, - autoraise=True, - ) - - thread = Thread(target=open_browser_thread) - thread.daemon = True - thread.start() - - @Slot() - def on_actionTestInformation_triggered(self): - if not self.d_f1_win.isVisible(): - self.d_f1_win.show() - - def on_openRecentFile(self): - action = self.sender() - if action: - self.reload_test_set_file(action.data()) - - def on_buttLogFilePath_clicked(self): - - if self.editLogFilePath.text() != "": - initialPath = os.path.dirname(self.editLogFilePath.text()) - elif self.testFile: - initialPath = os.path.dirname(self.testFile) - else: - initialPath = None - - fileName, _ = QFileDialog.getSaveFileName( - self, "Path to log file", initialPath, "Log Files (*.log);;All Files (*)" - ) - if fileName: - self.editLogFilePath.setText(fileName) - self.on_configLog_changed() - - def on_selectDeselectAll(self): - state = self.checkSelect.checkState() - self.disconnect_signals() - try: - if state == Qt.Checked: - self.treeTests.checkUncheckAll(self.test_service, True) - elif state == Qt.Unchecked: - self.treeTests.checkUncheckAll(self.test_service, False) - finally: - self.reconnect_signals() - - def on_testChecked(self, item, index): - self.checkSelect.setCheckState(Qt.PartiallyChecked) - self.disconnect_signals() - try: - self.treeTests.updateTreeCheckState(item, self.test_service) - finally: - self.reconnect_signals() - - @Slot() - def on_testSelectionChanged(self): - items = self.treeTests.selectedItems() - if len(items) > 0: - doc = items[0].doc - tmstmp = items[0].timestamp() - self.textEditTestDoc.setText("" + items[0].name + ":
") - if str(doc) != "": - self.textEditTestDoc.append(doc) - if tmstmp > 0: - text = self.textLog.toPlainText() - index = text.find(f"@@{tmstmp}@@") - if index != -1: - cursor = self.textLog.textCursor() - cursor.setPosition(index) - self.textLog.setTextCursor(cursor) - # obtain the vertical position of the cursor - block_number = cursor.blockNumber() - scrollbar = self.textLog.verticalScrollBar() - # Position the vert scrollbar to the right location - scrollbar.setValue(block_number) - - # Content of the F1 window is updated - self.update_f1_window(items[0]) - if self.d_f1_win.isVisible(): - self.d_f1_win.raise_() - - # When the test is selected, an attemp to move the log edit - # to the test is done. - # rmk: it has no effect when test is running. It is due to QPlainTextEdit - # limitations - if tmstmp > 0: - # Place the cursor at the begining of the text - cursor = self.textLog.textCursor() - cursor.movePosition(QTextCursor.Start) - self.textLog.setTextCursor(cursor) - # Find the timestamp - if self.textLog.find(f"@@{tmstmp}@@"): - cursor = self.textLog.textCursor() - ln = cursor.block().blockNumber() - # Move the scrollbar to the text - self.textLog.verticalScrollBar().setValue(ln) - cursor.clearSelection() - self.textLog.setTextCursor(cursor) - - def on_testItemDblClicked(self, item, col): - isBrkpointCol = item.setBreakpointIfCol(col) - if isBrkpointCol: - if item.isBreakpoint(): - self.test_service.add_breakpoint(item.id) - else: - self.test_service.del_breakpoint(item.id) - return - - s = sys.platform - - if (self.logFileName is not None) and os.access(self.logFileName, os.R_OK): - ln = tm.line_number("@@{}@@".format(item.timestamp()), self.logFileName) - - if ln > 0: - os.system("{} -g {}:{} &".format("code", self.logFileName, ln + 1)) - - def on_spacePressed(self): - item = self.treeTests.currentItem() - add_breakpoint = item.setBreakpoint() - if add_breakpoint: - self.test_service.add_breakpoint(item.id) - else: - self.test_service.del_breakpoint(item.id) - - def on_F1Pressed(self): - item = self.treeTests.currentItem() - self.update_f1_window(item) - self.d_f1_win.setVisible(True) - - # @Slot() - def on_breakpoint(self): - self._test_paused = True - self.startPauseTimer() - - def on_checkFoldChanged(self): - self.disconnect_signals() - try: - if self.checkFold.checkState() != Qt.Unchecked: - self.treeTests.foldAll(True) - self.checkFold.setCheckState(Qt.Checked) - else: - self.treeTests.foldAll(False) - finally: - self.reconnect_signals() - - def on_itemFoldChanged(self): - self.disconnect_signals() - try: - self.checkFold.setCheckState(Qt.PartiallyChecked) - finally: - self.reconnect_signals() - - def on_buttClearLog_clicked(self): - self.textLog.clear() - - def on_buttGoBottom_clicked(self): - self.textLog.moveCursor(QtGui.QTextCursor.End) - self.textLog.ensureCursorVisible() - - def on_configLog_changed(self): - prefs.settings.log_file = self.editLogFilePath.text() - - def on_configLogSaved_changed(self): - prefs.settings.log_file_saved = self.buttLogFileSaved.isChecked() - - def on_configLogNone_changed(self): - prefs.settings.log_file_saved = not self.buttLogFileNone.isChecked() - - def on_timerEvent(self): - text_to_append = [] - while not self.threads_queue.empty(): - text_to_append.append(self.threads_queue.get()) - - if len(text_to_append) > 0: - for t in text_to_append: - self.textLog.appendPlainText(t) - - if self.logFileHandler is not None: - self.logFileHandler.write(t + "\n") - self.logFileHandler.flush() - # os.fsync(self.logFileHandler) - - def on_timerBlinkEvent(self): - if self.buttBlink.current_color != "gray": - self.setBlinkGray() - elif self.treeTests.getGlobalSuccess(): - self.setBlinkGreen() - else: - self.setBlinkRed() - - def on_timerPause(self): - if self._test_paused: - icon = QtGui.QIcon() - if self.timerPause.state: - icon.addPixmap( - QtGui.QPixmap(icon_prefix() + "/pause2.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - else: - icon.addPixmap( - QtGui.QPixmap(icon_prefix() + "/pause.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.timerPause.state = not self.timerPause.state - self.actionStart_test.setIcon(icon) - - def on_timerCount(self): - secfromstart = self.start_time.secsTo(QDateTime.currentDateTime()) - self.label_runtime.setText( - "%02d:%02d:%02d" - % (secfromstart / 3600, (secfromstart / 60) % 60, secfromstart % 60) - ) - - def on_logToBeAppended(self, m): - self.textLog.moveCursor(QtGui.QTextCursor.End) - self.textLog.insertPlainText(m) - - def update_from_prefs(self): - self.hide_doc_pane() - self.hide_log_pane() - - def hide_doc_pane(self): - if prefs.settings.hide_doc_pane: - self.DocDockWidget.hide() - else: - self.DocDockWidget.show() - - def hide_log_pane(self): - if prefs.settings.hide_log_pane: - self.logDockWidget.hide() - else: - self.logDockWidget.show() - - def show_checkboxes(self, hidden=None): - if hidden: - h = hidden - else: - h = prefs.settings.show_checkboxes - if h: - # lab mode - if hasattr(self, "treeTests"): - self.disconnect_signals() - self.treeTests.addCheckBoxes() - self.reconnect_signals() - self.checkSelect.setEnabled(True) - else: - # production mode - if hasattr(self, "treeTests"): - self.treeTests.checkUncheckAll(self.test_service, True) - self.disconnect_signals() - self.treeTests.removeCheckBoxes() - self.reconnect_signals() - self.checkSelect.setDisabled(True) - - def addFileToRecent(self, filename): - files = prefs.settings.recent_files - - try: - files.remove(filename) - except ValueError: - pass - - files.insert(0, filename) - del files[MainWindow.MaxRecentFiles :] - - prefs.settings.recent_files = files - - for widget in QApplication.topLevelWidgets(): - if isinstance(widget, MainWindow): - widget.updateRecentFileActions() - - def updateRecentFileActions(self): - files = prefs.settings.recent_files - - numRecentFiles = min(len(files), MainWindow.MaxRecentFiles) - - for i in range(numRecentFiles): - text = "&%d %s" % (i + 1, self.strippedName(files[i])) - self.recentFileActs[i].setText(text) - self.recentFileActs[i].setData(files[i]) - self.recentFileActs[i].setVisible(True) - - for j in range(numRecentFiles, MainWindow.MaxRecentFiles): - self.recentFileActs[j].setVisible(False) - - self.separatorAct.setVisible((numRecentFiles > 0)) - - def update_f1_window(self, tree_item): - self.d_f1_win.ui.typeLineEdit.setText(tree_item.test_type) - self.d_f1_win.ui.sequenceFileNameLineEdit.setText(tree_item.seq_filename) - if tree_item.content is not None and tree_item.content != "": - self.d_f1_win.ui.TestContentEdit.setText(tree_item.content) - else: - self.d_f1_win.ui.TestContentEdit.setText("") - - def strippedName(self, fullFileName): - fname = os.path.basename(fullFileName) - fdir = os.path.dirname(fullFileName) - if len(fdir) > 30: - return os.path.join("... " + fdir[30:], fname) - else: - return fullFileName - - def defaults_for_process(self): - d = {} - - pp = prefs.settings.python_bin - if pp != "": - d["python_bin"] = pp - - pp = prefs.settings.lua_bin - if pp != "": - d["lua_bin"] = pp - - return d - - def loadTestSetFile(self, file_name): - """Load the tests: - return True if it succeeds, False otherwise. - """ - try: - if not file_name: - raise ETUMFileError("No file to load") - - file_name = os.path.abspath(file_name) - initial_dir = os.path.dirname(file_name) - - if not os.path.isdir(initial_dir): - raise ETUMFileError("Could not find %s directory" % (initial_dir)) - if not os.path.isfile(file_name): - raise ETUMFileError("Could not find %s file" % (file_name)) - - self.testFile = None - self.ts_controller = TestSetController() - self.test_service = TestControllerService(self.ts_controller) - self.test_proc = TestProcess( - file_name, - self.status_queue, - self.ts_controller, - self.config_files, - self.defines, - self.defaults_for_process(), - ) - self.test_proc.start() - while self.test_proc.is_alive(): - try: - if self.test_service.loaded(timeout=1.0): - break - except Empty: - self.test_service.clear() - - if not self.test_proc.is_alive(): - del self.test_proc - self.test_proc = None - del self.test_service - self.test_service = None - del self.ts_controller - self.ts_controller = None - - raise ETUMRuntimeError( - "Test could not be loaded (test process crashed for any reason)" - ) - - test_data = self.test_service.tree() - self.treeTests.clear() - self.treeTests.loadTestRecursively( - self.treeTests.invisibleRootItem(), test_data - ) - self.treeTests.setFoldDefault() - self.treeTests.updateTreeSkipState(self.test_service) - - self.checkSelect.setChecked(True) - self.testFile = file_name - test_dir = os.path.dirname(self.testFile) - - sys.path.append(test_dir) - self.statusBar().showMessage("Test file loaded", 10000) - self.textLog.set_test_dir(test_dir) - self.addFileToRecent(file_name) - self.setWindowTitle(self.mainWindowTitle + " - " + self.testFile) - self.actionStart_test.setEnabled(True) - self.actionRefresh_test.setEnabled(True) - - self.show_checkboxes() - return True - except: - self.statusBar().showMessage("No test file could be loaded", 10000) - self.treeTests.clear() - print(traceback.format_exc()) - return False - - def adaptInterfaceDuringTest(self): - try: - self.disconnect_signals() - # disable run and reload button - self.actionOpenTest.setDisabled(True) - self.actionExit.setDisabled(True) - icon = QtGui.QIcon() - icon.addPixmap( - QtGui.QPixmap(icon_prefix() + "/pause.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionStart_test.setIcon(icon) - self.actionStart_test.setText("Pause test") - self.actionPreferences.setDisabled(True) - self.actionRefresh_test.setDisabled(True) - self.actionShow_Results.setDisabled(True) - self.actionSave_report.setDisabled(True) - self.logSettingsBox.setDisabled(True) - self.actionStop_test.setEnabled(True) - if prefs.settings.show_checkboxes: - self._checklist = self.treeTests.getCheckList() - self.treeTests.removeCheckBoxes() - self.checkSelect.setDisabled(True) - self.checkFold.setDisabled(True) - self.timerBlink.setSingleShot(False) - self.timerBlink.setInterval(1000) - self.timerBlink.start() - self.setBlinkGreen() - self.treeTests.clearGlobalSuccess() - finally: - self._test_started = True - - def restoreInterfaceAfterTest(self): - try: - self.timerPause.stop() - self.timerBlink.stop() - # enable run and reload button - self.actionOpenTest.setEnabled(True) - self.actionExit.setEnabled(True) - icon = QtGui.QIcon() - icon.addPixmap( - QtGui.QPixmap(icon_prefix() + "/start.png"), - QtGui.QIcon.Normal, - QtGui.QIcon.Off, - ) - self.actionStart_test.setIcon(icon) - self.actionStart_test.setText("Start test") - self.actionPreferences.setEnabled(True) - self.actionRefresh_test.setEnabled(True) - self.actionStop_test.setDisabled(True) - self.actionShow_Results.setEnabled(True) - self.actionSave_report.setEnabled(True) - self.logSettingsBox.setEnabled(True) - if prefs.settings.show_checkboxes: - self.checkSelect.setEnabled(True) - self.treeTests.showCheckBoxes(self._checklist, self.test_service) - self.checkFold.setEnabled(True) - self.treeTests.setChildrenEnabled() - self.reconnect_signals() - if self.treeTests.getGlobalSuccess(): - self.setBlinkGreen() - else: - self.setBlinkRed() - finally: - self._test_started = False - - def redirectStdToTextLog(self, txtlog=None): - if txtlog is None: - stdio_redir.restore() - else: - stdio_redir.redirect(txtlog) - - def setBlinkGreen(self): - self.buttBlink.setIcon(self.iconBlinkGreen) - self.buttBlink.current_color = "green" - - def setBlinkRed(self): - self.buttBlink.setIcon(self.iconBlinkRed) - self.buttBlink.current_color = "red" - - def setBlinkGray(self): - self.buttBlink.setIcon(self.iconBlinkGray) - self.buttBlink.current_color = "gray" - - -def MainWin( - test_file=None, - config_files="", - run=False, - log_file="", - defines="", - report="", - report_type="", - report_pattern=[], - debug=False, -): - app = QApplication(sys.argv) - ui = MainWindow( - test_file, - config_files, - run, - log_file, - defines, - report, - report_type, - report_pattern, - debug, - ) - - ui.show() - sys.exit(app.exec_()) +import sys +import os +import subprocess +import traceback +import webbrowser +from time import sleep +from multiprocessing import Queue +from queue import Empty +from threading import Thread +import shutil +import ast + +# Qt +from PySide6 import QtGui, QtWidgets +from PySide6.QtGui import QAction, QShortcut, QIcon, QPixmap, QTextCursor, QDesktopServices, QTextCursor +from PySide6.QtCore import Slot, QUrl, Qt, QTimer, QDateTime + +from PySide6.QtWidgets import ( + QApplication, + QMainWindow, + QDialog, + QFileDialog, + QSizePolicy, +) + +ourPath = os.path.dirname(__file__) +sys.path.append(os.path.join(ourPath, "resources")) + +# user interfaces +from main_win.testium_core_win import Ui_MainWindow +from main_win.text_log import QTextLog +from main_win.about_win.about_win import Ui_About +from main_win.preference_win.preference_win import PrefWindow +from main_win.f1_win.d_f1_win import DialogF1 +from main_win.test_tree import QTestTree + +from main_win.test_run.thread_output import ThreadTestOutput +from lib.string_queue import StringQueue +from interpreter.process import TestProcess +from interpreter.utils.test_ctrl import TestSetController +from interpreter.utils.icons import icon_prefix + +from main_win.test_run.outlog import OutLog +from main_win.test_run.test_run import ThreadTestStatus +import interpreter.utils.settings as prefs +from lib.stdout_redirect import stdio_redir +import libs.testium as tm +from interpreter.utils.version import get_testium_version +from interpreter.utils.test_init import ( + env_init, + locate_report_file, +) +from lib.tum_except import ETUMFileError, ETUMRuntimeError +from main_win.test_controller_service import TestControllerService +from main_win.test_runner import TestRunner +from main_win.test_file_manager import TestFileManager + + +class MainWindow(QMainWindow, Ui_MainWindow): + MaxRecentFiles = 5 + + def __init__( + self, + test_file=None, + config_files="", + runandclose=False, + log_file="", + defines={}, + report="", + report_type="", + report_pattern=[], + debug=False, + ): + super().__init__() + self.setupUi(self) + self.textLog = self.create_text_log(self.frame1) + self.verticalLayout_2.addWidget(self.textLog) + + self._setup_icons() + + self.runandclose = runandclose + self.mainWindowTitle = self.windowTitle() + self.defines = defines + self.logFileName = log_file + self.reportFileName = report + self.report_type = report_type + self.report_pattern = report_pattern + self.config_files = config_files + self.recentFileActs = [] + self.debug = debug + self.test_proc = None + self.ts_controller = None + self.test_service = None + self.threadTestStatus = None + self._test_started = False + self._test_paused = False + self._signals_connected = False + + self.timer = QTimer() + self.timer.setSingleShot(False) + self.timer.stop() + self.timer.setInterval(100) + + self.timerBlink = QTimer() + self.timerBlink.setSingleShot(False) + self.timerBlink.stop() + self.timerBlink.setInterval(1000) + self.timerPause = QTimer() + self.timerPause.setSingleShot(False) + self.timerPause.stop() + self.timerPause.setInterval(500) + self.timerPause.state = False + self.iconBlinkGreen = QIcon() + self.iconBlinkGreen.addPixmap(QPixmap(icon_prefix() + "/green.png")) + self.iconBlinkRed = QIcon() + self.iconBlinkRed.addPixmap(QPixmap(icon_prefix() + "/red.png")) + self.iconBlinkGray = QIcon() + self.iconBlinkGray.addPixmap(QPixmap(icon_prefix() + "/gray.png")) + + self.threads_queue = Queue() + self.status_queue = Queue() + + # Managers + self.runner = TestRunner(self) + self.file_manager = TestFileManager(self) + + self.runner.set_blink_green() + + env_init() + + # Persistence + self.pref_win = PrefWindow(self) + + lastLog = prefs.settings.log_file + if self.logFileName == "": + self.editLogFilePath.setText(lastLog) + self.logFileName = lastLog + if prefs.settings.log_file_saved: + self.buttLogFileSaved.setChecked(True) + else: + if not os.path.isabs(self.logFileName): + self.logFileName = os.path.join(os.getcwd(), self.logFileName) + self.buttLogFileSaved.setChecked(True) + self.editLogFilePath.setText(self.logFileName) + + geo_settings = prefs.settings.value( + prefs.SettingsItem("geometry", bytearray), bytearray() + ) + if geo_settings: + self.restoreGeometry(geo_settings) + + state_settings = prefs.settings.value( + prefs.SettingsItem("state", bytearray), bytearray() + ) + if state_settings: + self.restoreState(state_settings) + + self.actionStart_test.setDisabled(True) + self.actionShow_Results.setDisabled(True) + self.actionSave_report.setDisabled(True) + + self.create_tree() + + self.shorcut_stop = QShortcut( + Qt.Key_Space, + self.treeTests, + context=Qt.WidgetShortcut, + activated=self.on_spacePressed, + ) + self.shorcut_f1 = QShortcut( + Qt.Key_F1, + self.treeTests, + context=Qt.WidgetShortcut, + activated=self.on_F1Pressed, + ) + + self.actionRefresh_test.setDisabled(True) + + # Signal connections + self.buttLogFilePath.pressed.connect(self.on_buttLogFilePath_clicked) + self.buttClearLog.pressed.connect(self.on_buttClearLog_clicked) + self.buttGoBottom.pressed.connect(self.on_buttGoBottom_clicked) + self.editLogFilePath.editingFinished.connect(self.on_configLog_changed) + self.buttLogFileSaved.toggled.connect(self.on_configLogSaved_changed) + self.buttLogFileNone.toggled.connect(self.on_configLogNone_changed) + self.timer.timeout.connect(self.runner.on_timer_event) + self.timerBlink.timeout.connect(self.runner.on_timer_blink) + self.timerBlink.timeout.connect(self.runner.on_timer_count) + self.timerPause.timeout.connect(self.runner.on_timer_pause) + self.treeTests.itemSelectionChanged.connect(self.on_testSelectionChanged) + if prefs.settings.dbl_click_enabled: + self.treeTests.setExpandsOnDoubleClick(False) + self.treeTests.itemDoubleClicked.connect(self.on_testItemDblClicked) + else: + self.treeTests.setExpandsOnDoubleClick(True) + QApplication.instance().lastWindowClosed.connect(self.on_exiting) + + self.prefs_apply_font() + self.prefs_apply_font_size() + + # Recent files menu + for i in range(MainWindow.MaxRecentFiles): + self.recentFileActs.append( + QAction(self, visible=False, triggered=self.file_manager.on_open_recent_file) + ) + self.separatorAct = self.menuFile.addSeparator() + for i in range(MainWindow.MaxRecentFiles): + self.menuFile.addAction(self.recentFileActs[i]) + self.file_manager.update_recent_file_actions() + + # Secondary windows + self.d_about_win = QDialog() + self.about_win = Ui_About() + self.about_win.setupUi(self.d_about_win) + self.about_win.labelVersion.setText("testium - " + get_testium_version()) + self.about_win.labelCesUnitVersion.setText("") + self.d_about_win.setModal(True) + + self.d_f1_win = DialogF1(self) + + self.stream = StringQueue() + stdio_redir.redirect(self.stream) + self.threadOutput = ThreadTestOutput(self.stream, self.threads_queue) + self.threadOutput.start() + + self.out_log = OutLog() + self.out_log.logToBeAppended.connect(self.on_logToBeAppended) + self.redirectStdToTextLog(self.out_log) + self.testFile = test_file + + self.threadTestStatus = ThreadTestStatus(self.status_queue, debug=self.debug) + self.threadTestStatus.start() + + self.update_from_prefs() + + self.reportFileName = locate_report_file(self.reportFileName) + + last_files = prefs.settings.recent_files + ret = False + if test_file != "": + if not os.path.isabs(test_file): + test_file = os.path.join(os.getcwd(), test_file) + if os.path.isfile(test_file): + ret = self.file_manager.load(test_file) + elif (len(last_files) > 0) and os.path.isfile(last_files[0]): + ret = self.file_manager.load(last_files[0]) + + if ret: + self.file_loaded_at_startup() + + self.threadTestStatus.testSetIsFinished.connect(self.runner.on_run_finished) + self.threadTestStatus.statusToBeUpdated.connect(self.treeTests.updateStatus) + self.reconnect_signals() + + if runandclose: + self.on_actionStart_test_triggered() + + def _setup_icons(self): + icons = { + self.buttClearLog: "edit-clear", + self.buttGoBottom: "go-bottom", + self.actionOpenTest: "document-open", + self.actionSave_report: "document-save", + self.actionStart_test: "start", + self.actionStop_test: "stop", + self.actionAbout_testium: "about", + self.actionExit: "exit", + self.actionRefresh_test: "view-refresh", + self.actionShow_Results: "results", + self.actionHelp: "help", + self.actionPreferences: "settings", + self.actionTestInformation: "info", + } + for widget, name in icons.items(): + icon = QtGui.QIcon() + icon.addPixmap( + QtGui.QPixmap(icon_prefix() + f"/{name}.png"), + QtGui.QIcon.Normal, + QtGui.QIcon.Off, + ) + widget.setIcon(icon) + + def create_text_log(self, parent): + return QTextLog(parent) + + def create_tree(self): + self.treeTests = QTestTree(self.widget) + self.treeTests.setEnabled(True) + sizePolicy = QSizePolicy(QSizePolicy.Minimum, QSizePolicy.Expanding) + sizePolicy.setHorizontalStretch(0) + sizePolicy.setVerticalStretch(0) + sizePolicy.setHeightForWidth(self.treeTests.sizePolicy().hasHeightForWidth()) + self.treeTests.setSizePolicy(sizePolicy) + self.treeTests.breakpoint.connect(self.on_breakpoint) + self.verticalLayout.addWidget(self.treeTests) + + def remove_tree(self): + self.verticalLayout.removeWidget(self.treeTests) + del self.treeTests + self.treeTests = None + + def file_loaded_at_startup(self): + modeSlider_value = prefs.settings.show_checkboxes + if modeSlider_value: + checkList = prefs.settings.value(prefs.SettingsItem("checkList", list), []) + if checkList is not None: + if len(checkList) == self.treeTests.getItemCount(): + self.treeTests.restoreCheckList(checkList, self.test_service) + else: + tm.print_info( + "The number of tests has changed. Test box states are not restored." + ) + foldList = prefs.settings.value(prefs.SettingsItem("foldList", list), []) + if foldList: + if len(foldList) == self.treeTests.getItemCount(): + self.checkFold.setCheckState(Qt.PartiallyChecked) + self.treeTests.restoreFoldList(foldList) + + def disconnect_signals(self): + if self._signals_connected: + self.checkSelect.stateChanged.disconnect() + self.treeTests.itemChanged.disconnect() + self.checkFold.stateChanged.disconnect() + self.treeTests.itemCollapsed.disconnect() + self.treeTests.itemExpanded.disconnect() + self._signals_connected = False + + def reconnect_signals(self): + if not self._signals_connected: + self.checkSelect.stateChanged.connect(self.on_selectDeselectAll) + self.treeTests.itemChanged.connect(self.on_testChecked) + self.checkFold.stateChanged.connect(self.on_checkFoldChanged) + self.treeTests.itemCollapsed.connect(self.on_itemFoldChanged) + self.treeTests.itemExpanded.connect(self.on_itemFoldChanged) + self._signals_connected = True + + def prefs_apply_font(self): + f = self.textLog.font() + f.fromString(prefs.settings.log_font) + self.textLog.setFont(f) + + def prefs_apply_font_size(self): + f = self.textLog.font() + f.setPointSize(prefs.settings.log_font_size) + self.textLog.setFont(f) + + def save_settings(self): + prefs.settings.set_value( + prefs.SettingsItem("geometry", bytearray), bytearray(self.saveGeometry()) + ) + prefs.settings.set_value( + prefs.SettingsItem("state", bytearray), bytearray(self.saveState()) + ) + prefs.settings.set_value( + prefs.SettingsItem("checkList", list), self.treeTests.getCheckList() + ) + prefs.settings.set_value( + prefs.SettingsItem("foldList", list), self.treeTests.getFoldList() + ) + self.treeTests.saveSizes() + prefs.settings.sync() + + def on_exiting(self): + if not self._test_started: + self.save_settings() + self.file_manager.clear_process() + self.threadTestStatus.stop() + self.threadOutput.stop() + self.threadOutput.wait() + self.threadTestStatus.wait() + + def show_checkboxes(self, hidden=None): + if hidden: + h = hidden + else: + h = prefs.settings.show_checkboxes + if h: + if hasattr(self, "treeTests"): + self.disconnect_signals() + self.treeTests.addCheckBoxes() + self.reconnect_signals() + self.checkSelect.setEnabled(True) + else: + if hasattr(self, "treeTests"): + self.treeTests.checkUncheckAll(self.test_service, True) + self.disconnect_signals() + self.treeTests.removeCheckBoxes() + self.reconnect_signals() + self.checkSelect.setDisabled(True) + + def update_from_prefs(self): + self.hide_doc_pane() + self.hide_log_pane() + + def hide_doc_pane(self): + if prefs.settings.hide_doc_pane: + self.DocDockWidget.hide() + else: + self.DocDockWidget.show() + + def hide_log_pane(self): + if prefs.settings.hide_log_pane: + self.logDockWidget.hide() + else: + self.logDockWidget.show() + + def update_f1_window(self, tree_item): + self.d_f1_win.ui.typeLineEdit.setText(tree_item.test_type) + self.d_f1_win.ui.sequenceFileNameLineEdit.setText(tree_item.seq_filename) + if tree_item.content is not None and tree_item.content != "": + self.d_f1_win.ui.TestContentEdit.setText(tree_item.content) + else: + self.d_f1_win.ui.TestContentEdit.setText("") + + def _stripped_name(self, fullFileName): + fname = os.path.basename(fullFileName) + fdir = os.path.dirname(fullFileName) + if len(fdir) > 30: + return os.path.join("... " + fdir[30:], fname) + else: + return fullFileName + + def redirectStdToTextLog(self, txtlog=None): + if txtlog is None: + stdio_redir.restore() + else: + stdio_redir.redirect(txtlog) + + # --- Qt Slots (thin delegates) --- + + @Slot() + def on_actionOpenTest_triggered(self): + self.file_manager.on_open_test() + + @Slot() + def on_actionStart_test_triggered(self): + self.runner.on_start_test() + + def on_runFinished(self): + self.runner.on_run_finished() + + @Slot() + def on_actionStop_test_triggered(self): + self.runner.on_stop_test() + + def on_breakpoint(self): + self.runner.on_breakpoint() + + @Slot() + def on_actionExit_triggered(self): + self.close() + + @Slot() + def on_actionAbout_testium_triggered(self): + self.d_about_win.setVisible(True) + + @Slot() + def on_actionPreferences_triggered(self): + result = self.pref_win.exec() + if result == QDialog.Accepted: + self.update_from_prefs() + if self.pref_win.isChanged(prefs.settings.SettingsShowCheckboxes): + self.show_checkboxes() + if self.pref_win.isChanged(prefs.settings.SettingsDblClickEnabled): + if prefs.settings.dbl_click_enabled: + self.treeTests.itemDoubleClicked.connect(self.on_testItemDblClicked) + else: + self.treeTests.itemDoubleClicked.disconnect() + if self.pref_win.isChanged(prefs.settings.SettingsLogFont): + self.prefs_apply_font() + if self.pref_win.isChanged(prefs.settings.SettingsLogFontSize): + self.prefs_apply_font_size() + + @Slot() + def on_actionRefresh_test_triggered(self): + self.on_exiting() + args = [] + if not hasattr(sys, "frozen"): + args += [sys.executable] + args += [sys.argv[0]] + if len(self.defines) > 0: + for k, v in self.defines.items(): + try: + val = ast.literal_eval(v) + except: + val = v + args += ["-d", f"{k}={val}"] + if (self.testFile is not None) and (isinstance(self.testFile, str)): + args += [self.testFile] + os.execv(sys.executable, args) + + @Slot() + def on_actionSave_report_triggered(self): + if self.testFile: + initialPath = os.path.dirname(self.testFile) + else: + initialPath = None + fileName, _ = QFileDialog.getSaveFileName( + self, "Path to Log file", initialPath, "Log Files (*.log);;All Files (*)" + ) + if fileName: + shutil.copy(self.logFileName, fileName) + + @Slot() + def on_actionShow_Results_triggered(self): + s = sys.platform + self.statusBar().showMessage( + "Opening the logfile (" + s + "): " + self.logFileName, 100000 + ) + QDesktopServices.openUrl(QUrl.fromLocalFile(self.logFileName)) + + @Slot() + def on_actionHelp_triggered(self): + self.webbrowser_open() + + def webbrowser_open(self): + def open_browser_thread(): + webbrowser.open( + "https://git.beafrancois.fr/v-and-v/testium/src/branch/main/doc/manual/testium_manual.pdf", + new=2, + autoraise=True, + ) + thread = Thread(target=open_browser_thread) + thread.daemon = True + thread.start() + + @Slot() + def on_actionTestInformation_triggered(self): + if not self.d_f1_win.isVisible(): + self.d_f1_win.show() + + def on_buttLogFilePath_clicked(self): + if self.editLogFilePath.text() != "": + initialPath = os.path.dirname(self.editLogFilePath.text()) + elif self.testFile: + initialPath = os.path.dirname(self.testFile) + else: + initialPath = None + fileName, _ = QFileDialog.getSaveFileName( + self, "Path to log file", initialPath, "Log Files (*.log);;All Files (*)" + ) + if fileName: + self.editLogFilePath.setText(fileName) + self.on_configLog_changed() + + def on_selectDeselectAll(self): + state = self.checkSelect.checkState() + self.disconnect_signals() + try: + if state == Qt.Checked: + self.treeTests.checkUncheckAll(self.test_service, True) + elif state == Qt.Unchecked: + self.treeTests.checkUncheckAll(self.test_service, False) + finally: + self.reconnect_signals() + + def on_testChecked(self, item, index): + self.checkSelect.setCheckState(Qt.PartiallyChecked) + self.disconnect_signals() + try: + self.treeTests.updateTreeCheckState(item, self.test_service) + finally: + self.reconnect_signals() + + @Slot() + def on_testSelectionChanged(self): + items = self.treeTests.selectedItems() + if len(items) > 0: + doc = items[0].doc + tmstmp = items[0].timestamp() + self.textEditTestDoc.setText("" + items[0].name + ":
") + if str(doc) != "": + self.textEditTestDoc.append(doc) + if tmstmp > 0: + text = self.textLog.toPlainText() + index = text.find(f"@@{tmstmp}@@") + if index != -1: + cursor = self.textLog.textCursor() + cursor.setPosition(index) + self.textLog.setTextCursor(cursor) + block_number = cursor.blockNumber() + scrollbar = self.textLog.verticalScrollBar() + scrollbar.setValue(block_number) + + self.update_f1_window(items[0]) + if self.d_f1_win.isVisible(): + self.d_f1_win.raise_() + + if tmstmp > 0: + cursor = self.textLog.textCursor() + cursor.movePosition(QTextCursor.Start) + self.textLog.setTextCursor(cursor) + if self.textLog.find(f"@@{tmstmp}@@"): + cursor = self.textLog.textCursor() + ln = cursor.block().blockNumber() + self.textLog.verticalScrollBar().setValue(ln) + cursor.clearSelection() + self.textLog.setTextCursor(cursor) + + def on_testItemDblClicked(self, item, col): + isBrkpointCol = item.setBreakpointIfCol(col) + if isBrkpointCol: + if item.isBreakpoint(): + self.test_service.add_breakpoint(item.id) + else: + self.test_service.del_breakpoint(item.id) + return + if (self.logFileName is not None) and os.access(self.logFileName, os.R_OK): + ln = tm.line_number("@@{}@@".format(item.timestamp()), self.logFileName) + if ln > 0: + os.system("{} -g {}:{} &".format("code", self.logFileName, ln + 1)) + + def on_spacePressed(self): + item = self.treeTests.currentItem() + add_breakpoint = item.setBreakpoint() + if add_breakpoint: + self.test_service.add_breakpoint(item.id) + else: + self.test_service.del_breakpoint(item.id) + + def on_F1Pressed(self): + item = self.treeTests.currentItem() + self.update_f1_window(item) + self.d_f1_win.setVisible(True) + + def on_checkFoldChanged(self): + self.disconnect_signals() + try: + if self.checkFold.checkState() != Qt.Unchecked: + self.treeTests.foldAll(True) + self.checkFold.setCheckState(Qt.Checked) + else: + self.treeTests.foldAll(False) + finally: + self.reconnect_signals() + + def on_itemFoldChanged(self): + self.disconnect_signals() + try: + self.checkFold.setCheckState(Qt.PartiallyChecked) + finally: + self.reconnect_signals() + + def on_buttClearLog_clicked(self): + self.textLog.clear() + + def on_buttGoBottom_clicked(self): + self.textLog.moveCursor(QtGui.QTextCursor.End) + self.textLog.ensureCursorVisible() + + def on_configLog_changed(self): + prefs.settings.log_file = self.editLogFilePath.text() + + def on_configLogSaved_changed(self): + prefs.settings.log_file_saved = self.buttLogFileSaved.isChecked() + + def on_configLogNone_changed(self): + prefs.settings.log_file_saved = not self.buttLogFileNone.isChecked() + + def on_logToBeAppended(self, m): + self.textLog.moveCursor(QtGui.QTextCursor.End) + self.textLog.insertPlainText(m) + + # --- Blink delegates (kept for backward compatibility with treeTests signal) --- + + def setBlinkGreen(self): + self.runner.set_blink_green() + + def setBlinkRed(self): + self.runner.set_blink_red() + + def setBlinkGray(self): + self.runner.set_blink_gray() + + +def MainWin( + test_file=None, + config_files="", + run=False, + log_file="", + defines="", + report="", + report_type="", + report_pattern=[], + debug=False, +): + app = QApplication(sys.argv) + ui = MainWindow( + test_file, + config_files, + run, + log_file, + defines, + report, + report_type, + report_pattern, + debug, + ) + + ui.show() + sys.exit(app.exec_())