Henrik Lemke
8ed1a4315f
- eco.utilities.mainthread: a watchdog interrupts the main thread (SIGUSR2) while a cell blocks, so Qt windows (camera viewers, plots, the desktop workbench) stay live during umv/scans/sleep; call_soon/on_main for GUI work from other threads. Camera viewers default to in-process again. - eco.utilities.task_queue: named queues (queues[name], %queue magic, scans.*(queue=...)) replacing scan_queue.py; every append is test-run (names, signatures, scan build, limits); claims block moves of a running task's adjustables from elsewhere (forced() overrides); output capture, recording and an output window (queue_output_qt). - eco.utilities.user_input: input() from a worker asks on the main thread; input_timeout(escalate=False) takes the default at once inside a queue. - scan: start_time %M -> %m bug fixed; GUI calls in scan callbacks routed to the main thread.
132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
"""Tests for eco.utilities.user_input: questions from other threads are shown
|
|
on the main thread, and unattended (queue) threads take timed defaults at
|
|
once."""
|
|
|
|
import builtins
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from eco.utilities import mainthread, user_input
|
|
|
|
|
|
@pytest.fixture
|
|
def heartbeat():
|
|
mainthread.uninstall()
|
|
hb = mainthread.install(scope="always") # no IPython shell under pytest
|
|
yield hb
|
|
mainthread.uninstall()
|
|
|
|
|
|
def in_thread(func):
|
|
result = {}
|
|
|
|
def run():
|
|
try:
|
|
result["value"] = func()
|
|
except BaseException as exc:
|
|
result["error"] = exc
|
|
|
|
thread = threading.Thread(target=run)
|
|
thread.start()
|
|
return thread, result
|
|
|
|
|
|
def wait_while_alive(thread, seconds=5):
|
|
t0 = time.monotonic()
|
|
while thread.is_alive() and time.monotonic() - t0 < seconds:
|
|
time.sleep(0.02) # main thread "busy": the heartbeat runs the dispatch
|
|
|
|
|
|
def test_input_timeout_in_an_unattended_thread_takes_the_default_at_once():
|
|
def asker():
|
|
with user_input.unattended("queue 'test'"):
|
|
t0 = time.monotonic()
|
|
with pytest.raises(user_input.InputTimeout):
|
|
user_input.input_timeout("Change back? (y/n)", timeout=10)
|
|
return time.monotonic() - t0
|
|
|
|
thread, result = in_thread(asker)
|
|
thread.join(5)
|
|
assert result["value"] < 0.5
|
|
|
|
|
|
def test_input_timeout_error_is_an_inputimeout_timeout():
|
|
inputimeout = pytest.importorskip("inputimeout")
|
|
assert issubclass(user_input.InputTimeout, inputimeout.TimeoutOccurred)
|
|
|
|
|
|
def test_question_from_a_worker_is_answered_on_the_main_thread(heartbeat, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake_channel(question):
|
|
seen["main"] = threading.current_thread() is threading.main_thread()
|
|
seen["origin"] = question.origin
|
|
return "42"
|
|
|
|
monkeypatch.setattr(user_input, "_ask_here", fake_channel)
|
|
|
|
def asker():
|
|
with user_input.unattended("queue 'test', task #1"):
|
|
return user_input.ask("value?")
|
|
|
|
thread, result = in_thread(asker)
|
|
wait_while_alive(thread)
|
|
assert result == {"value": "42"}
|
|
assert seen == {"main": True, "origin": "queue 'test', task #1"}
|
|
|
|
|
|
def test_cancelled_question_raises_in_the_asking_thread(heartbeat, monkeypatch):
|
|
def cancel(question):
|
|
raise user_input.QuestionCancelled(question.prompt)
|
|
|
|
monkeypatch.setattr(user_input, "_ask_here", cancel)
|
|
thread, result = in_thread(lambda: user_input.ask("sure?"))
|
|
wait_while_alive(thread)
|
|
assert isinstance(result["error"], user_input.QuestionCancelled)
|
|
|
|
|
|
def test_routed_input_from_a_worker_goes_through_ask(heartbeat, monkeypatch):
|
|
monkeypatch.setattr(user_input, "_ask_here", lambda question: f"answer to {question.prompt}")
|
|
monkeypatch.setattr(builtins, "input", user_input._routed_input)
|
|
thread, result = in_thread(lambda: input("(y/n)"))
|
|
wait_while_alive(thread)
|
|
assert result == {"value": "answer to (y/n)"}
|
|
|
|
|
|
def test_without_any_channel_the_question_waits_for_answer(heartbeat, monkeypatch):
|
|
monkeypatch.setattr(user_input, "_ask_here", lambda question: NotImplemented)
|
|
thread, result = in_thread(lambda: user_input.ask("which edge?"))
|
|
t0 = time.monotonic()
|
|
while not user_input.pending_questions() and time.monotonic() - t0 < 5:
|
|
time.sleep(0.02)
|
|
assert [q.prompt for q in user_input.pending_questions()] == ["which edge?"]
|
|
assert user_input.answer(0.512) is True
|
|
thread.join(5)
|
|
assert result == {"value": "0.512"}
|
|
assert user_input.pending_questions() == []
|
|
|
|
|
|
def test_qt_dialog_channel(heartbeat, monkeypatch):
|
|
QtWidgets = pytest.importorskip("qtpy.QtWidgets")
|
|
from qtpy import QtCore
|
|
|
|
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
|
mainthread.install_qt()
|
|
monkeypatch.setattr(user_input, "_terminal_shell", lambda: None)
|
|
|
|
def answer_dialog():
|
|
dialog = QtWidgets.QApplication.activeModalWidget()
|
|
if dialog is None:
|
|
QtCore.QTimer.singleShot(50, answer_dialog)
|
|
return
|
|
dialog.setTextValue("from the dialog")
|
|
dialog.accept()
|
|
|
|
QtCore.QTimer.singleShot(200, answer_dialog)
|
|
thread, result = in_thread(lambda: user_input.ask("name?"))
|
|
wait_while_alive(thread)
|
|
assert result == {"value": "from the dialog"}
|
|
del app
|