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 <noreply@anthropic.com>
This commit is contained in:
2026-04-18 13:32:45 +02:00
parent 3ed73d93a7
commit a353511f64
3 changed files with 1109 additions and 1171 deletions

View File

@@ -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)

View File

@@ -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"

View File

@@ -4,7 +4,6 @@ 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
@@ -39,7 +38,6 @@ 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
@@ -53,6 +51,9 @@ from interpreter.utils.test_init import (
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):
@@ -75,103 +76,10 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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._setup_icons()
self.runandclose = runandclose
# Var init
self.mainWindowTitle = self.windowTitle()
self.logFileHandler = None
self.defines = defines
self.logFileName = log_file
self.reportFileName = report
@@ -208,11 +116,16 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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()
# Managers
self.runner = TestRunner(self)
self.file_manager = TestFileManager(self)
self.runner.set_blink_green()
env_init()
# Persistence
@@ -242,15 +155,12 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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,
@@ -264,20 +174,19 @@ class MainWindow(QMainWindow, Ui_MainWindow):
activated=self.on_F1Pressed,
)
# Main Window items modifications
self.actionRefresh_test.setDisabled(True)
# Connection of the handlers
# 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.on_timerEvent)
self.timerBlink.timeout.connect(self.on_timerBlinkEvent)
self.timerBlink.timeout.connect(self.on_timerCount)
self.timerPause.timeout.connect(self.on_timerPause)
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)
@@ -289,29 +198,27 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.prefs_apply_font()
self.prefs_apply_font_size()
# Recent files
# Recent files menu
for i in range(MainWindow.MaxRecentFiles):
self.recentFileActs.append(
QAction(self, visible=False, triggered=self.on_openRecentFile)
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.updateRecentFileActions()
self.file_manager.update_recent_file_actions()
# A propos
# 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)
# F1 window
self.d_f1_win = DialogF1(self)
self.stream = StringQueue() # stream used to log output
self.stream = StringQueue()
stdio_redir.redirect(self.stream)
self.threadOutput = ThreadTestOutput(self.stream, self.threads_queue)
self.threadOutput.start()
@@ -326,36 +233,55 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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)
ret = self.file_manager.load(test_file)
elif (len(last_files) > 0) and os.path.isfile(last_files[0]):
ret = self.loadTestSetFile(last_files[0])
ret = self.file_manager.load(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.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):
textLog = QTextLog(parent)
return textLog
return QTextLog(parent)
def create_tree(self):
self.treeTests = QTestTree(self.widget)
@@ -375,21 +301,15 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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():
@@ -398,7 +318,6 @@ class MainWindow(QMainWindow, Ui_MainWindow):
def disconnect_signals(self):
if self._signals_connected:
# disconnect the GUI
self.checkSelect.stateChanged.disconnect()
self.treeTests.itemChanged.disconnect()
self.checkFold.stateChanged.disconnect()
@@ -408,7 +327,6 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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)
@@ -426,143 +344,6 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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())
@@ -579,19 +360,96 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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.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)
@@ -620,29 +478,23 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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 (*)"
)
@@ -668,7 +520,6 @@ class MainWindow(QMainWindow, Ui_MainWindow):
new=2,
autoraise=True,
)
thread = Thread(target=open_browser_thread)
thread.daemon = True
thread.start()
@@ -678,20 +529,13 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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 (*)"
)
@@ -734,31 +578,21 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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)
@@ -771,12 +605,8 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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))
@@ -793,11 +623,6 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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:
@@ -832,315 +657,20 @@ class MainWindow(QMainWindow, Ui_MainWindow):
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)
# --- Blink delegates (kept for backward compatibility with treeTests signal) ---
def setBlinkGreen(self):
self.buttBlink.setIcon(self.iconBlinkGreen)
self.buttBlink.current_color = "green"
self.runner.set_blink_green()
def setBlinkRed(self):
self.buttBlink.setIcon(self.iconBlinkRed)
self.buttBlink.current_color = "red"
self.runner.set_blink_red()
def setBlinkGray(self):
self.buttBlink.setIcon(self.iconBlinkGray)
self.buttBlink.current_color = "gray"
self.runner.set_blink_gray()
def MainWin(