v-and-v/testium
2
0
Fork 0
Code Issues 5 Pull Requests 1 Actions Packages Projects Releases 7 Wiki Activity

Fix dialog subprocesses: centralize Qt env setup and extract base class

Browse Source 
- Add dialog_env.py service: forces QT_QPA_PLATFORM=xcb on Linux so Qt
  doesn't crash under Wayland in spawned subprocesses
- Use QMessageBox instance (instead of static methods) for msg/question
  dialogs so WindowStaysOnTopHint can be set, making them visible
- Add TestItemDialogBase with _run_dialog/_run_dialog_with_result/_cleanup_process,
  removing duplicated subprocess launch/poll/terminate logic from all 7 dialog items
- Reduce terminate() join timeout from 2s to 0.2s across all dialog items

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Foué
2026-04-19 20:01:56 +02:00
parent c72176d029
commit 2b7678c39e
17 changed files with 429 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/testium/interpreter/test_items/dialog_choices_files/choices_dialog.py
View File

@@ -185,7 +185,9 @@ def main(args, conn=None):
SettingsApplication = "testium_choices_dlg_" + args[0] SettingsApplication = "testium_choices_dlg_" + args[0]
SettingsLastChoices = "last_choice" SettingsLastChoices = "last_choice"
success = True success = True
app = QApplication() from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = ChoicesDialog() d = ChoicesDialog()
d.setFixedSize(800, 600) d.setFixedSize(800, 600)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

15
src/testium/interpreter/test_items/dialog_env.py Normal file
View File

@@ -0,0 +1,15 @@
"""Qt platform environment setup for dialog subprocesses.
Call setup() at the start of every dialog subprocess main() function
to ensure the correct Qt platform plugin is selected.
"""
import sys
import os
def setup():
"""Configure the Qt environment for dialog subprocess usage."""
if sys.platform.startswith('linux'):
# On Linux/Wayland, force X11 (via XWayland) to avoid crashes
# when Qt is initialized inside a multiprocessing subprocess.
os.environ['QT_QPA_PLATFORM'] = 'xcb'

5
src/testium/interpreter/test_items/dialog_image_files/dialog_image.py
View File

@@ -1,5 +1,4 @@
import sys import sys
import os
from PySide6.QtCore import (Qt) from PySide6.QtCore import (Qt)
from PySide6.QtWidgets import (QApplication, QDialog) from PySide6.QtWidgets import (QApplication, QDialog)
@@ -18,7 +17,9 @@ class TestDialogWindow(QDialog, dialog_image_win.Ui_Dialog):
def main(args, conn): def main(args, conn):
success = True success = True
app = QApplication(args) from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow() d = TestDialogWindow()
d.setFixedSize(700,600) d.setFixedSize(700,600)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

73
src/testium/interpreter/test_items/dialog_msg_files/msg_dialog.py
View File

@@ -1,36 +1,37 @@
import sys import sys
import os from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QDialog) from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import (Qt) from PySide6.QtCore import Qt
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args):
def main(args): from interpreter.test_items import dialog_env
app = QApplication(sys.argv) dialog_env.setup()
reply = QMessageBox.information(None, args[0], args[1], QMessageBox.Ok) app = QApplication(['testium'])
msg = QMessageBox()
if hasattr(sys, "frozen"): msg.setWindowFlags(Qt.WindowStaysOnTopHint)
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug. msg.setWindowTitle(args[0])
class dummyStream: msg.setText(args[1])
''' dummyStream behaves like a stream but does nothing. ''' msg.setIcon(QMessageBox.Information)
def __init__(self): pass msg.setStandardButtons(QMessageBox.Ok)
def write(self,data): pass msg.exec()
def read(self,data): pass
def flush(self): pass if hasattr(sys, "frozen"):
def close(self): pass class dummyStream:
def __init__(self): pass
# and now redirect all default streams to this dummyStream: def write(self, data): pass
sys.stdout = dummyStream() def read(self, data): pass
sys.stderr = dummyStream() def flush(self): pass
sys.stdin = dummyStream() def close(self): pass
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream() sys.stdout = dummyStream()
sys.__stdin__ = dummyStream() sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
if __name__ == '__main__': sys.__stderr__ = dummyStream()
main(sys.argv[1:]) sys.__stdin__ = dummyStream()
if __name__ == '__main__':
main(sys.argv[1:])

4
src/testium/interpreter/test_items/dialog_note_files/test_dialog.py
View File

@@ -14,7 +14,9 @@ class TestDialogWindow(QDialog, dialog_note_win.Ui_Dialog):
def main(args, conn=None): def main(args, conn=None):
success = True success = True
app = QApplication(args) from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow() d = TestDialogWindow()
d.setFixedSize(387,224) d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

71
src/testium/interpreter/test_items/dialog_question_files/question_dialog.py
View File

@@ -1,32 +1,39 @@
import sys import sys
import os from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QDialog) from PySide6.QtWidgets import (QApplication, QMessageBox)
from PySide6.QtCore import (Qt) from PySide6.QtCore import Qt
from PySide6.QtWidgets import QMessageBox
from multiprocessing import freeze_support
def main(args, conn):
def main(args, conn): try:
app = QApplication(sys.argv) from interpreter.test_items import dialog_env
reply = QMessageBox.question(None, args[0], args[1], QMessageBox.Yes|QMessageBox.No) dialog_env.setup()
app = QApplication(['testium'])
conn.send(reply) msg = QMessageBox()
conn.close() msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.setWindowTitle(args[0])
if hasattr(sys, "frozen"): msg.setText(args[1])
#all standard streams are replaced by dummy one to avoid cx_freeze flushing bug. msg.setIcon(QMessageBox.Question)
class dummyStream: msg.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
''' dummyStream behaves like a stream but does nothing. ''' reply = msg.exec()
def __init__(self): pass conn.send(reply)
def write(self,data): pass except Exception as e:
def read(self,data): pass print(f"dialog_question error: {e}", file=sys.stderr)
def flush(self): pass finally:
def close(self): pass conn.close()
# and now redirect all default streams to this dummyStream: if hasattr(sys, "frozen"):
sys.stdout = dummyStream() class dummyStream:
sys.stderr = dummyStream() def __init__(self): pass
sys.stdin = dummyStream() def write(self, data): pass
sys.__stdout__ = dummyStream() def read(self, data): pass
sys.__stderr__ = dummyStream() def flush(self): pass
sys.__stdin__ = dummyStream() def close(self): pass
sys.stdout = dummyStream()
sys.stderr = dummyStream()
sys.stdin = dummyStream()
sys.__stdout__ = dummyStream()
sys.__stderr__ = dummyStream()
sys.__stdin__ = dummyStream()

4
src/testium/interpreter/test_items/dialog_sleep_files/dialog_sleep.py
View File

@@ -39,7 +39,9 @@ class DialogSleepWindow(QDialog, dialog_sleep_win.Ui_SleepDialogWindow):
def main(args, conn=None): def main(args, conn=None):
success = True success = True
app = QApplication(sys.argv) from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = DialogSleepWindow() d = DialogSleepWindow()
d.setFixedSize(379,129) d.setFixedSize(379,129)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

4
src/testium/interpreter/test_items/dialog_value_files/test_dialog.py
View File

@@ -15,7 +15,9 @@ class TestDialogWindow(QDialog, dialog_value_win.Ui_Dialog):
def main(args, conn=None): def main(args, conn=None):
success = True success = True
app = QApplication(args) from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestDialogWindow() d = TestDialogWindow()
d.setFixedSize(387,224) d.setFixedSize(387,224)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)

80
src/testium/interpreter/test_items/test_item_choices_dialog.py
View File

@@ -1,43 +1,37 @@
from multiprocessing import Process, Pipe from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item import TestItem, test_run from interpreter.test_items.dialog_choices_files import choices_dialog
from interpreter.test_items.test_result import TestResult, TestValue from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.dialog_choices_files import choices_dialog from interpreter.utils.constants import TestItemType as cst
import libs.testium as tm from lib.tum_except import item_load_context
from lib.tum_except import ETUMSyntaxError, item_load_context import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst
class TestItemChoicesDialog(TestItemDialogBase):
class TestItemChoicesDialog(TestItem): def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""): self._name = cst.TYPE_CHOICES_DLG.item_name
self._name = cst.TYPE_CHOICES_DLG.item_name super().__init__(dict_item, parent, status_queue, filename=filename)
super().__init__(dict_item, parent, status_queue, filename=filename) self._type = cst.TYPE_CHOICES_DLG
self._type = cst.TYPE_CHOICES_DLG self.is_container = False
self.is_container = False with item_load_context(self.cmd(), self.name(), self.seqFilename()):
with item_load_context(self.cmd(), self.name(), self.seqFilename()): self._question = self._prms.getParam("question", required=True)
self._question = self._prms.getParam("question", required=True) self._choices = self._prms.getParam("choices", required=True)
self._choices = self._prms.getParam("choices", required=True) self._default_icon = self._prms.getParam("icon", required=False, default=None)
self._default_icon = self._prms.getParam("icon", required=False, default=None)
@test_run
@test_run def execute(self):
def execute(self): q = self._prms.expanse(self._question)
q = self._prms.expanse(self._question) choices = self._prms.expanse(self._choices)
choices = self._prms.expanse(self._choices) icon = self._prms.expanse(self._default_icon)
icon = self._prms.expanse(self._default_icon) result = self._run_dialog_with_result(choices_dialog.main, [self.name(), q, choices, icon])
parent_conn, child_conn = Pipe() if result is None:
p = Process( self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
target=choices_dialog.main, args=([self.name(), q, choices, icon], child_conn) return
) val, succ = result
p.start() self.result.value = val
val, succ = parent_conn.recv() if succ:
p.join() tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
self.result.value = val else:
tm.delgd("cs_" + self._name)
if succ: self.result.set(TestValue.FAILURE, str(val))
# The result of the test item is put into the global dict
tm.setgd("cs_" + self._name, val)
self.result.set(TestValue.SUCCESS, str(val))
else:
tm.delgd("cs_" + self._name)
self.result.set(TestValue.FAILURE, str(val))

43
src/testium/interpreter/test_items/test_item_dialog_base.py Normal file
View File

@@ -0,0 +1,43 @@
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import TestItem
class TestItemDialogBase(TestItem):
"""Base class for test items that launch a Qt dialog in a subprocess."""
def _cleanup_process(self, p):
if p.is_alive():
p.terminate()
p.join(timeout=0.2)
if p.is_alive():
p.kill()
p.join()
def _run_dialog(self, target, args):
"""Launch target(args) in a subprocess with no return value.
Returns the subprocess exit code.
"""
p = Process(target=target, args=(args,))
p.start()
while p.is_alive() and not self._is_stopped:
p.join(timeout=0.5)
self._cleanup_process(p)
return p.exitcode
def _run_dialog_with_result(self, target, args):
"""Launch target(args, child_conn) in a subprocess and return what it sends.
Returns the received value, or None if stopped or if the subprocess crashed.
"""
parent_conn, child_conn = Pipe()
p = Process(target=target, args=(args, child_conn))
p.start()
result = None
while p.is_alive() and not self._is_stopped:
if parent_conn.poll(0.5):
result = parent_conn.recv()
break
self._cleanup_process(p)
return result

106
src/testium/interpreter/test_items/test_item_image_dialog.py
View File

@@ -1,66 +1,40 @@
import os import os
import sys
from multiprocessing import Process, Pipe from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_item import TestItem, test_run from interpreter.test_items.dialog_image_files import dialog_image
from interpreter.test_items.test_result import TestResult, TestValue from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.dialog_image_files import dialog_image from interpreter.utils.constants import TestItemType as cst
import libs.testium as tm from lib.tum_except import item_load_context
from interpreter.utils.constants import TestItemType as cst import libs.testium as tm
from lib.tum_except import ETUMSyntaxError, item_load_context
class TestItemImageDialog(TestItemDialogBase):
class TestItemImageDialog(TestItem): """dialog_image item usage.
"""dialog_image item usage. dialog_image name: Nice image, question: could you press the red button, filename: img.jpg
dialog_image name: Nice image, question: could you press the red button, filename: img.jpg """
""" def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
self._name = cst.TYPE_IMAGE_DLG.item_name
def __init__(self, dict_item, parent=None, status_queue=None, filename=""): super().__init__(dict_item, parent, status_queue, filename=filename)
self._name = cst.TYPE_IMAGE_DLG.item_name self._type = cst.TYPE_IMAGE_DLG
super().__init__(dict_item, parent, status_queue, filename=filename) self.is_container = False
self._type = cst.TYPE_IMAGE_DLG with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.is_container = False self._question = self._prms.getParam("question", required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()): self._filename = self._prms.getParam("filename", required=True)
self._question = self._prms.getParam("question", required=True)
self._filename = self._prms.getParam("filename", required=True) @test_run
def execute(self):
@test_run q = self._prms.expanse(self._question)
def execute(self): image_path = self._prms.expanse(self._filename)
ourpath = __file__ print("Image Displayed:\n" + q + "\n" + image_path)
test_file = os.path.join( if not os.path.isfile(image_path):
os.path.dirname(ourpath), "dialog_image_files", "dialog_image.py" image_path = os.path.normpath(
) os.path.join(tm.gd("test_directory"), image_path)
)
q = self._prms.expanse(self._question) succ = self._run_dialog_with_result(dialog_image.main, [self.name(), q, image_path])
image_path = self._prms.expanse(self._filename) if succ is None:
print("Image Displayed:\n" + q + "\n" + image_path) self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
if not os.path.isfile(image_path): elif succ:
image_path = os.path.normpath( self.result.set(TestValue.SUCCESS)
os.path.join(tm.gd("test_directory"), image_path) else:
) self.result.set(TestValue.FAILURE)
parent_conn, child_conn = Pipe()
p = Process(
target=dialog_image.main, args=([self.name(), q, image_path], child_conn)
)
p.start()
succ = parent_conn.recv()
p.join()
if succ:
self.result.set(TestValue.SUCCESS)
else:
self.result.set(TestValue.FAILURE)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__ == "__main__":
p = Process(target=test_dialog.main, args=(["bob", "bab"],))
p.start()
p.join()

81
src/testium/interpreter/test_items/test_item_msg_dialog.py
View File

@@ -1,49 +1,32 @@
import os import os
import sys import sys
from multiprocessing import Process, Pipe
from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_item import (TestItem, test_run) from interpreter.test_items.test_result import TestValue
from interpreter.test_items.test_result import (TestValue) from interpreter.test_items.dialog_msg_files import msg_dialog
from interpreter.test_items.dialog_msg_files import msg_dialog from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.utils.constants import TestItemType as cst from interpreter.utils.constants import TestItemType as cst
from lib.tum_except import ETUMSyntaxError, item_load_context from lib.tum_except import item_load_context
class TestItemMsgDialog(TestItem):
"""dialog_message item usage. class TestItemMsgDialog(TestItemDialogBase):
dialog_message name: Nice message, question: Open the door and press OK """dialog_message item usage.
""" dialog_message name: Nice message, question: Open the door and press OK
def __init__(self, dict_item, parent = None, status_queue=None, filename=""): """
self._name = cst.TYPE_MESSAGE_DLG.item_name def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
super().__init__(dict_item, parent, status_queue, filename=filename) self._name = cst.TYPE_MESSAGE_DLG.item_name
self._type = cst.TYPE_MESSAGE_DLG super().__init__(dict_item, parent, status_queue, filename=filename)
self.is_container = False self._type = cst.TYPE_MESSAGE_DLG
with item_load_context(self.cmd(), self.name(), self.seqFilename()): self.is_container = False
self._question = self._prms.getParam('question', required=True) with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True)
@test_run
def execute(self): @test_run
ourpath = __file__ def execute(self):
test_file = os.path.join(os.path.dirname(ourpath), q = self._prms.expanse(self._question)
'dialog_msg_files', print("Message Displayed:\n" + q)
'msg_dialog.py') exitcode = self._run_dialog(msg_dialog.main, [self.name(), q])
if exitcode == 0:
q = self._prms.expanse(self._question) self.result.set(TestValue.SUCCESS)
print("Message Displayed:\n" + q) else:
parent_conn, child_conn = Pipe() self.result.set(TestValue.FAILURE, f"Dialog subprocess exited with code {exitcode}")
p=Process(target=msg_dialog.main,
args=([self.name(), q],))
p.start()
p.join()
self.result.set(TestValue.SUCCESS)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=msg_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()

95
src/testium/interpreter/test_items/test_item_note_dialog.py
View File

@@ -1,57 +1,38 @@
import os from interpreter.test_items.test_item import test_run
import sys from interpreter.test_items.test_result import TestValue
from multiprocessing import Process, Pipe from interpreter.test_items.dialog_note_files import test_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item import (TestItem, test_run) from interpreter.utils.constants import TestItemType as cst
from interpreter.test_items.test_result import (TestResult, TestValue) from lib.tum_except import item_load_context
from interpreter.test_items.dialog_note_files import test_dialog import libs.testium as tm
from lib.tum_except import ETUMSyntaxError, item_load_context
import libs.testium as tm
from interpreter.utils.constants import TestItemType as cst class TestItemNoteDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
class TestItemNoteDialog(TestItem): self._name = cst.TYPE_NOTE_DLG.item_name
def __init__(self, dict_item, parent = None, status_queue=None, filename=""): super().__init__(dict_item, parent, status_queue, filename=filename)
self._name = cst.TYPE_NOTE_DLG.item_name self._type = cst.TYPE_NOTE_DLG
super().__init__(dict_item, parent, status_queue, filename=filename) self.is_container = False
self._type = cst.TYPE_NOTE_DLG with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.is_container = False self._question = self._prms.getParam('question', required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True) @test_run
def execute(self):
@test_run q = self._prms.expanse(self._question)
def execute(self): print("Question:\n" + q)
ourpath = __file__ result = self._run_dialog_with_result(test_dialog.main, [self.name(), q])
test_file = os.path.join(os.path.dirname(ourpath), if result is None:
'dialog_note_files', self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
'test_dialog.py') return
val, succ = result
q = self._prms.expanse(self._question) tm.setgd(self.name(), val)
print("Question:\n" + q) print("\n" + ("-" * 80) + "\n")
parent_conn, child_conn = Pipe() print("- Test note\n")
p=Process(target=test_dialog.main, args=([self.name(), q],child_conn)) print("-" * 80 + "\n")
p.start() print(val)
val, succ = parent_conn.recv() print("-" * 80 + "\n")
p.join() self.result.reported = {'note': val}
tm.setgd(self.name(), val) if succ:
print("\n" + ("-" * 80) + "\n") self.result.set(TestValue.SUCCESS, val)
print("- Test note\n") else:
print("-" * 80 + "\n") self.result.set(TestValue.FAILURE, val)
print(val)
print("-" * 80 + "\n")
self.result.reported = {'note': val}
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()

93
src/testium/interpreter/test_items/test_item_question_dialog.py
View File

@@ -1,57 +1,36 @@
import os from PySide6.QtWidgets import QMessageBox
import sys
from multiprocessing import Process, Pipe from interpreter.test_items.test_item import test_run
from interpreter.test_items.test_result import TestValue
from PySide6.QtWidgets import QMessageBox from interpreter.test_items.dialog_question_files import question_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item import (TestItem, test_run) from interpreter.utils.constants import TestItemType as cst
from interpreter.test_items.test_result import (TestResult, TestValue) from lib.tum_except import item_load_context
from interpreter.test_items.dialog_question_files import question_dialog
from lib.tum_except import ETUMSyntaxError, item_load_context
from interpreter.utils.constants import TestItemType as cst class TestItemQuestionDialog(TestItemDialogBase):
"""dialog_question item usage.
class TestItemQuestionDialog(TestItem): dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel"
"""dialog_question item usage. """
dialog_question name: Nice question, question: "If OK, press OK, If not, press cancel" def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
""" self._name = cst.TYPE_QUESTION_DLG.item_name
def __init__(self, dict_item, parent = None, status_queue=None, filename=""): super().__init__(dict_item, parent, status_queue, filename=filename)
self._name = cst.TYPE_QUESTION_DLG.item_name self._type = cst.TYPE_QUESTION_DLG
super().__init__(dict_item, parent, status_queue, filename=filename) self.is_container = False
self._type = cst.TYPE_QUESTION_DLG with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.is_container = False self._question = self._prms.getParam('question', required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self._question = self._prms.getParam('question', required=True) @test_run
def execute(self):
@test_run q = self._prms.expanse(self._question)
def execute(self): print('Question asked:\n' + q + '\n')
ourpath = __file__ succ = self._run_dialog_with_result(question_dialog.main, [self.name(), q])
test_file = os.path.join(os.path.dirname(ourpath), if succ is None:
'dialog_question_files', self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
'question_dialog.py') return
if succ == QMessageBox.Yes:
q = self._prms.expanse(self._question) self.result.set(TestValue.SUCCESS)
print('Question asked:\n' + q + '\n') print('Answer: YES\n')
parent_conn, child_conn = Pipe() else:
p=Process(target=question_dialog.main, self.result.set(TestValue.FAILURE)
args=([self.name(), q],child_conn)) print('Answer: NO\n')
p.start()
succ = parent_conn.recv()
p.join()
if succ == QMessageBox.Yes:
self.result.set(TestValue.SUCCESS)
print('Answer: YES\n')
else:
self.result.set(TestValue.FAILURE)
print('Answer: NO\n')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()

123
src/testium/interpreter/test_items/test_item_tested_references.py
View File

@@ -1,72 +1,51 @@
import os from interpreter.test_items.test_item import test_run
import sys from interpreter.test_items.test_result import TestValue
from multiprocessing import Process, Pipe from interpreter.test_items.tested_references_files import tested_refs_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item import (TestItem, test_run) from interpreter.utils.constants import TestItemType as cst
from interpreter.test_items.test_result import (TestResult, TestValue) from lib.tum_except import item_load_context
from interpreter.test_items.tested_references_files import tested_refs_dialog import libs.testium as tm
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError, item_load_context
from interpreter.utils.constants import TestItemType as cst class TestItemTestedRefsDialog(TestItemDialogBase):
def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
class TestItemTestedRefsDialog(TestItem): self._name = cst.TYPE_REFERENCE_DLG.item_name
def __init__(self, dict_item, parent=None, status_queue=None, filename=""): super().__init__(dict_item, parent, status_queue, filename=filename)
self._name = cst.TYPE_REFERENCE_DLG.item_name self._type = cst.TYPE_REFERENCE_DLG
super().__init__(dict_item, parent, status_queue, filename=filename) self.is_container = False
self._type = cst.TYPE_REFERENCE_DLG with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.is_container = False self._question = self._prms.getParam('question', required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()): self._init_values = self._prms.getParamAll('reference', required=False, processed=True)
self._question = self._prms.getParam('question', required=True)
self._init_values = self._prms.getParamAll('reference', required=False, processed=True) @test_run
def execute(self):
@test_run q = self._prms.expanse(self._question)
def execute(self): init_values = ','.join(self._init_values)
ourpath=__file__ result = self._run_dialog_with_result(tested_refs_dialog.main, [self.name(), q, init_values])
test_file=os.path.join(os.path.dirname(ourpath), if result is None:
'tested_references_files', self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
'tested_refs_dialog.py') return
val, succ = result
q=self._prms.expanse(self._question)
parent_conn, child_conn=Pipe() titems = []
init_values=','.join(self._init_values) if len(val) > 0:
p=Process(target=tested_refs_dialog.main, i = 0
args=([self.name(), q, init_values], for sitem in val.split(','):
child_conn)) titem = {}
p.start() telems = sitem.split('/')
val, succ=parent_conn.recv() titem['reference'] = telems[0]
p.join() titem['revision'] = telems[1]
titem['serial'] = telems[2]
titems=[] print("Identification:\n" + str(titem))
if len(val) > 0: titems.append(titem)
i = 0 self.result.reported = {'reference_{}'.format(i): titem}
for sitem in val.split(','): i += 1
titem={} self.result.value = titems
telems=sitem.split('/') tm.setgd('tested_items', titems)
titem['reference']=telems[0] if len(val) > 0:
titem['revision']=telems[1] if succ:
titem['serial']=telems[2] self.result.set(TestValue.SUCCESS, val)
print("Identification:\n" + str(titem)) else:
titems.append(titem) self.result.set(TestValue.FAILURE, val)
self.result.reported = {'reference_{}'.format(i): titem} else:
i = i + 1 self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
self.result.value = titems
tm.setgd('tested_items', titems)
if len(val) > 0:
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__ == '__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()

105
src/testium/interpreter/test_items/test_item_value_dialog.py
View File

@@ -1,62 +1,43 @@
import os from interpreter.test_items.test_item import test_run
import sys from interpreter.test_items.test_result import TestValue
from multiprocessing import Process, Pipe from interpreter.test_items.dialog_value_files import test_dialog
from interpreter.test_items.test_item_dialog_base import TestItemDialogBase
from interpreter.test_items.test_item import (TestItem, test_run) from interpreter.utils.constants import TestItemType as cst
from interpreter.test_items.test_result import (TestResult, TestValue) from lib.tum_except import item_load_context
from interpreter.test_items.dialog_value_files import test_dialog import libs.testium as tm
import libs.testium as tm
from lib.tum_except import ETUMSyntaxError, item_load_context
from interpreter.utils.constants import TestItemType as cst class TestItemValueDialog(TestItemDialogBase):
"""dialog_value item usage.
class TestItemValueDialog(TestItem): dialog_value name: Enter value, question: "Which value did you measure?"
"""dialog_value item usage. """
dialog_value name: Enter value, question: "Which value did you measure?" def __init__(self, dict_item, parent=None, status_queue=None, filename=""):
""" self._name = cst.TYPE_VALUE_DLG.item_name
def __init__(self, dict_item, parent = None, status_queue=None, filename=""): super().__init__(dict_item, parent, status_queue, filename=filename)
self._name = cst.TYPE_VALUE_DLG.item_name self._type = cst.TYPE_VALUE_DLG
super().__init__(dict_item, parent, status_queue, filename=filename) self.is_container = False
self._type = cst.TYPE_VALUE_DLG with item_load_context(self.cmd(), self.name(), self.seqFilename()):
self.is_container = False self._question = self._prms.getParam('question', required=True)
with item_load_context(self.cmd(), self.name(), self.seqFilename()): self._default = self._prms.getParam('default', '')
self._question = self._prms.getParam('question', required=True)
self._default = self._prms.getParam('default', '') @test_run
def execute(self):
@test_run q = self._prms.expanse(self._question)
def execute(self): d = self._prms.expanse(self._default)
ourpath = __file__ print("Question:\n" + q)
test_file = os.path.join(os.path.dirname(ourpath), result = self._run_dialog_with_result(test_dialog.main, [self.name(), q, d])
'dialog_value_files', if result is None:
'test_dialog.py') self.result.set(TestValue.FAILURE, "Dialog subprocess exited without returning a result")
return
q = self._prms.expanse(self._question) val, succ = result
d = self._prms.expanse(self._default) tm.setgd(self.name(), val)
print("Question:\n" + q) print("Answer: " + val)
parent_conn, child_conn = Pipe() if len(val) > 0:
p=Process(target=test_dialog.main, args=([self.name(), q, d],child_conn)) self.result.reported = {'question': q, 'answer': val}
p.start() self.result.value = val
val, succ = parent_conn.recv() if succ:
p.join() self.result.set(TestValue.SUCCESS, val)
tm.setgd(self.name(), val) else:
print("Answer: " + val) self.result.set(TestValue.FAILURE, val)
if len(val) > 0: else:
self.result.reported = {'question': q, 'answer': val} self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
self.result.value = val
if succ:
self.result.set(TestValue.SUCCESS, val)
else:
self.result.set(TestValue.FAILURE, val)
else:
self.result.set(TestValue.FAILURE, 'The dialog did not return any value')
def mypath():
if hasattr(sys, "frozen"):
return os.path.dirname(sys.executable)
return os.path.dirname(__file__)
from multiprocessing import Process
if __name__=='__main__':
p=Process(target=test_dialog.main, args=(['bob', 'bab'],))
p.start()
p.join()

5
src/testium/interpreter/test_items/tested_references_files/tested_refs_dialog.py
View File

@@ -1,5 +1,4 @@
import sys import sys
import os
from multiprocessing import freeze_support from multiprocessing import freeze_support
from PySide6.QtWidgets import (QApplication, QDialog, QTableWidgetItem) from PySide6.QtWidgets import (QApplication, QDialog, QTableWidgetItem)
@@ -20,7 +19,9 @@ def main(args, conn=None):
SettingsApplication = 'testium_ref_item' SettingsApplication = 'testium_ref_item'
SettingsLastReference = 'lastReference' SettingsLastReference = 'lastReference'
success = True success = True
app = QApplication(args) from interpreter.test_items import dialog_env
dialog_env.setup()
app = QApplication(['testium'])
d = TestedRefsWindow() d = TestedRefsWindow()
d.setFixedSize(481,386) d.setFixedSize(481,386)
d.setWindowFlags(Qt.WindowStaysOnTopHint) d.setWindowFlags(Qt.WindowStaysOnTopHint)