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.
502 lines
16 KiB
Python
502 lines
16 KiB
Python
"""Tests for eco.utilities.task_queue: named queues of statements/callables
|
|
on a worker thread, their test run, and the %queue magic's parsing."""
|
|
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from eco.utilities import mainthread, task_queue, user_input
|
|
from eco.utilities.task_queue import QueueTestError, TaskQueue
|
|
|
|
|
|
@pytest.fixture
|
|
def ns():
|
|
return {"__builtins__": __builtins__}
|
|
|
|
|
|
@pytest.fixture
|
|
def queue(ns):
|
|
q = TaskQueue("test", namespace=ns)
|
|
yield q
|
|
q.stop()
|
|
q.close()
|
|
|
|
|
|
def finish(q, timeout=5):
|
|
q.wait(timeout=timeout)
|
|
|
|
|
|
def test_statements_run_in_order_in_the_shared_namespace(queue, ns):
|
|
ns["log"] = []
|
|
queue.append("log.append(1)")
|
|
queue.append("x = 41")
|
|
queue.append("log.append(x + 1)")
|
|
finish(queue)
|
|
assert ns["log"] == [1, 42]
|
|
assert ns["x"] == 41
|
|
assert [t.state for t in queue.history] == ["done", "done", "done"]
|
|
|
|
|
|
def test_a_variable_changed_meanwhile_is_seen_when_the_task_runs(queue, ns):
|
|
ns["gate"] = threading.Event()
|
|
ns["start"] = 0
|
|
queue.append("gate.wait(5)")
|
|
queue.append("seen = start")
|
|
ns["start"] = 7 # changed "at the prompt" before the task runs
|
|
ns["gate"].set()
|
|
finish(queue)
|
|
assert ns["seen"] == 7
|
|
|
|
|
|
def test_lone_expression_result_is_kept(queue):
|
|
task = queue.append("6 * 7")
|
|
assert task.wait(5) == 42
|
|
|
|
|
|
def test_callable_tasks(queue):
|
|
calls = []
|
|
task = queue.append(calls.append, "hello")
|
|
task.wait(5)
|
|
assert calls == ["hello"]
|
|
|
|
|
|
def test_typo_in_a_name_is_rejected_with_a_suggestion(queue, ns):
|
|
ns["scans"] = object()
|
|
with pytest.raises(QueueTestError) as info:
|
|
queue.append("scnas.ascan(1, 2)")
|
|
assert "did you mean 'scans'" in str(info.value)
|
|
assert queue.pending == []
|
|
|
|
|
|
def test_typo_in_an_attribute_is_rejected(queue, ns):
|
|
class Mono:
|
|
energy = None
|
|
|
|
ns["mono"] = Mono()
|
|
with pytest.raises(QueueTestError) as info:
|
|
queue.append("mono.enrgy")
|
|
assert "did you mean 'energy'" in str(info.value)
|
|
|
|
|
|
def test_wrong_arguments_are_rejected(queue, ns):
|
|
ns["f"] = lambda a, b: None
|
|
with pytest.raises(QueueTestError) as info:
|
|
queue.append("f(1, 2, 3)")
|
|
assert "too many positional arguments" in str(info.value)
|
|
|
|
|
|
def test_names_assigned_by_earlier_tasks_count_as_defined(queue, ns):
|
|
queue.pause()
|
|
queue.append("edge = 0.5")
|
|
queue.append("result = edge * 2") # `edge` does not exist yet, but will
|
|
queue.resume()
|
|
finish(queue)
|
|
assert ns["result"] == 1.0
|
|
|
|
|
|
def test_properties_are_not_evaluated_by_the_test(queue, ns):
|
|
reads = []
|
|
|
|
class Device:
|
|
@property
|
|
def value(self):
|
|
reads.append(1)
|
|
return 1
|
|
|
|
ns["dev"] = Device()
|
|
queue.pause()
|
|
queue.append("v = dev.value")
|
|
assert reads == []
|
|
|
|
|
|
def test_move_outside_limits_is_rejected(queue, ns):
|
|
class Motor:
|
|
def get_limits(self):
|
|
return (-1, 1)
|
|
|
|
def get_current_value(self):
|
|
return 0.5
|
|
|
|
def mv(self, value):
|
|
pass
|
|
|
|
def mvr(self, value):
|
|
pass
|
|
|
|
ns["m"] = Motor()
|
|
with pytest.raises(QueueTestError):
|
|
queue.append("m.mv(5)")
|
|
with pytest.raises(QueueTestError):
|
|
queue.append("m.mvr(0.8)") # 0.5 + 0.8 > 1
|
|
queue.pause()
|
|
queue.append("m.mv(0.2)")
|
|
|
|
|
|
def test_test_run_reports_pending_problems(queue, ns):
|
|
queue.pause()
|
|
queue.append("y = 1")
|
|
queue.append("missing_name + 1", test=False)
|
|
report = queue.test_run(verbose=False)
|
|
assert list(report) == [queue.pending[1]]
|
|
|
|
|
|
def test_failure_pauses_the_queue(queue, ns):
|
|
ns["log"] = []
|
|
queue.append("1 / 0")
|
|
queue.append("log.append('after')")
|
|
time.sleep(0.5)
|
|
assert queue.paused
|
|
assert queue.history[0].state == "error"
|
|
assert ns["log"] == []
|
|
queue.resume()
|
|
finish(queue)
|
|
assert ns["log"] == ["after"]
|
|
|
|
|
|
def test_pause_cancel_and_repeat_last(queue, ns):
|
|
ns["gate"] = threading.Event()
|
|
ns["log"] = []
|
|
queue.append("gate.wait(5)")
|
|
queue.append("log.append('a')")
|
|
queue.append("log.append('b')")
|
|
time.sleep(0.2)
|
|
dropped = queue.cancel_after_running_task()
|
|
assert len(dropped) == 2
|
|
queue.pause_after_running_task()
|
|
queue.append("log.append('c')")
|
|
queue.repeat_last(N=2)
|
|
ns["gate"].set()
|
|
time.sleep(0.3)
|
|
assert ns["log"] == [] # paused after the running task
|
|
queue.resume()
|
|
finish(queue)
|
|
assert ns["log"] == ["c", "c", "c"]
|
|
|
|
|
|
def test_stop_running_task_interrupts_plain_python(queue, ns):
|
|
task = queue.append("while True: pass")
|
|
time.sleep(0.3)
|
|
queue.stop_running_task()
|
|
with pytest.raises(KeyboardInterrupt):
|
|
task.wait(5)
|
|
assert queue.paused
|
|
|
|
|
|
def test_tasks_run_unattended_and_know_their_task(queue, ns):
|
|
ns["task_queue"] = task_queue
|
|
ns["user_input"] = user_input
|
|
task = queue.append("(task_queue.current_task().id, user_input.unattended_label())")
|
|
task_id, label = task.wait(5)
|
|
assert task_id == task.id
|
|
assert label == task.label
|
|
|
|
|
|
def test_input_timeout_in_a_task_takes_the_default_at_once(queue, ns):
|
|
ns["user_input"] = user_input
|
|
queue.append(
|
|
"t0 = __import__('time').monotonic()\n"
|
|
"try:\n"
|
|
" user_input.input_timeout('go on? (y/n)', timeout=10)\n"
|
|
"except user_input.InputTimeout:\n"
|
|
" took = __import__('time').monotonic() - t0"
|
|
)
|
|
finish(queue)
|
|
assert ns["took"] < 0.5
|
|
|
|
|
|
def test_input_in_a_task_is_asked_on_the_main_thread(queue, ns, monkeypatch):
|
|
mainthread.uninstall()
|
|
mainthread.install(scope="always")
|
|
try:
|
|
monkeypatch.setattr(user_input, "_ask_here", lambda question: f"{question.origin}: yes")
|
|
ns["user_input"] = user_input
|
|
task = queue.append("answer = user_input.ask('sure?')")
|
|
t0 = time.monotonic()
|
|
while task.state != "done" and time.monotonic() - t0 < 5:
|
|
time.sleep(0.02)
|
|
assert ns["answer"] == f"{task.label}: yes"
|
|
finally:
|
|
mainthread.uninstall()
|
|
|
|
|
|
def test_description_of_a_running_task_reaches_its_scans(queue):
|
|
class FakeScan:
|
|
description = None
|
|
|
|
def set_description(self, text):
|
|
self.description = text
|
|
|
|
task = task_queue.Task("pass")
|
|
scan = FakeScan()
|
|
task.scans.append(scan)
|
|
task.description = "new title"
|
|
assert scan.description == "new title"
|
|
|
|
|
|
def test_magic_appends_and_runs_subcommands(monkeypatch):
|
|
q = TaskQueue("magic", namespace={"__builtins__": __builtins__})
|
|
monkeypatch.setitem(task_queue.queues, "magic", q)
|
|
try:
|
|
q.pause()
|
|
task_queue.queue_magic("-q magic x = 1")
|
|
task_queue.queue_magic("-q magic repeat_last 2")
|
|
assert [t.source for t in q.pending] == ["x = 1"] * 3
|
|
task_queue.queue_magic("-q magic cancel_after_running_task")
|
|
assert q.pending == []
|
|
task_queue.queue_magic("-q magic -- status = 3") # a statement, not the subcommand
|
|
assert [t.source for t in q.pending] == ["status = 3"]
|
|
finally:
|
|
q.stop()
|
|
q.close()
|
|
|
|
|
|
def test_magic_reports_a_failed_test_instead_of_raising(monkeypatch, capsys):
|
|
q = TaskQueue("magic2", namespace={"__builtins__": __builtins__})
|
|
monkeypatch.setitem(task_queue.queues, "magic2", q)
|
|
try:
|
|
task_queue.queue_magic("-q magic2 undefined_thing()")
|
|
assert "not queued" in capsys.readouterr().out
|
|
assert q.pending == []
|
|
finally:
|
|
q.close()
|
|
|
|
|
|
# -- scans in queues ------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def scans_setup():
|
|
"""A real Scans instance with in-memory adjustables and a fake counter --
|
|
nothing touches hardware or the filesystem."""
|
|
from eco.acquisition.scan import Scans
|
|
from eco.acquisition.utilities import Acquisition
|
|
from eco.elements.adjustable import DummyAdjustable
|
|
|
|
class FakeCounter:
|
|
name = "fake"
|
|
|
|
def __init__(self):
|
|
self.threads = []
|
|
|
|
def acquire(self, scan=None, Npulses=None, **kwargs):
|
|
self.threads.append(threading.current_thread().name)
|
|
return Acquisition(acquire=lambda: time.sleep(0.01), hold=False)
|
|
|
|
counter = FakeCounter()
|
|
scans = Scans(default_counters=[counter], name="scans")
|
|
mot = DummyAdjustable(name="mot", limits=[-2, 2])
|
|
mot2 = DummyAdjustable(name="mot2", limits=[-2, 2])
|
|
return scans, mot, mot2, counter
|
|
|
|
|
|
def test_scans_queue_keyword_appends_and_runs_on_the_worker(scans_setup, monkeypatch):
|
|
scans, mot, _, counter = scans_setup
|
|
q = TaskQueue("scan-test", namespace={"__builtins__": __builtins__})
|
|
monkeypatch.setitem(task_queue.queues, "scan-test", q)
|
|
try:
|
|
task = scans.ascan(mot, 0, 1, 2, 1, queue="scan-test", return_at_end=False)
|
|
assert isinstance(task, task_queue.Task)
|
|
scan = task.wait(20)
|
|
assert scan.adjustables == [mot]
|
|
assert task.scan is scan
|
|
assert set(counter.threads) == {"eco-queue-scan-test"}
|
|
assert mot.get_current_value() == 1
|
|
finally:
|
|
q.stop()
|
|
q.close()
|
|
|
|
|
|
def test_test_building_a_scan_moves_nothing_and_keeps_acquiring_scan(scans_setup, monkeypatch):
|
|
scans, mot, _, counter = scans_setup
|
|
before = scans.acquiring_scan
|
|
q = TaskQueue("scan-test2", namespace={"__builtins__": __builtins__, "scans": scans, "mot": mot})
|
|
monkeypatch.setitem(task_queue.queues, "scan-test2", q)
|
|
try:
|
|
q.pause()
|
|
q.append("scans.ascan(mot, 0, 1, 2, 1)")
|
|
assert counter.threads == [] # nothing acquired
|
|
assert mot.get_current_value() == 0 # nothing moved
|
|
assert scans.acquiring_scan is before
|
|
with pytest.raises(QueueTestError) as info:
|
|
q.append("scans.ascan(mot, 0, 5, 2, 1)")
|
|
assert "outside its limits (-2, 2)" in str(info.value)
|
|
finally:
|
|
q.stop()
|
|
q.close()
|
|
|
|
|
|
def test_a2scan_can_be_test_built(scans_setup):
|
|
scans, mot, mot2, _ = scans_setup
|
|
scan = scans.a2scan(mot, 0, 1, mot2, 0, 3, 2, 1, start_immediately=False)
|
|
from eco.acquisition.scan_tasks import check_scan_limits
|
|
|
|
problems = check_scan_limits(scan)
|
|
assert len(problems) == 1 and "mot2" in problems[0]
|
|
|
|
|
|
def test_queue_output_goes_to_the_configured_stream(queue, ns, monkeypatch):
|
|
import io
|
|
import sys
|
|
|
|
console = io.StringIO()
|
|
monkeypatch.setattr(sys, "stdout", sys.stdout)
|
|
monkeypatch.setattr(task_queue, "_output", {"stdout": None, "stderr": None})
|
|
monkeypatch.setattr(sys, "stderr", sys.stderr)
|
|
task_queue.send_output_to(console)
|
|
try:
|
|
queue.append("print('from the queue')")
|
|
finish(queue)
|
|
print("from the main thread")
|
|
assert console.getvalue() == "from the queue\n"
|
|
finally:
|
|
sys.stdout = sys.stdout._fallback
|
|
sys.stderr = sys.stderr._fallback
|
|
|
|
|
|
# -- claims: what a running task moves can't be moved from elsewhere ----------------
|
|
|
|
|
|
def test_a_tasks_motor_cannot_be_moved_from_the_prompt_unless_forced(queue, ns):
|
|
from eco.elements.adjustable import DummyAdjustable
|
|
|
|
mot = DummyAdjustable(name="mot")
|
|
ns["mot"] = mot
|
|
ns["gate"] = threading.Event()
|
|
task = queue.append("mot.set_target_value(1).wait(); gate.wait(5)")
|
|
t0 = time.monotonic()
|
|
while mot.get_current_value() != 1 and time.monotonic() - t0 < 5:
|
|
time.sleep(0.02)
|
|
with pytest.raises(task_queue.MoveBlocked) as info:
|
|
mot.set_target_value(5).wait()
|
|
message = str(info.value)
|
|
assert f"in use by {task.label}" in message and "with forced():" in message
|
|
assert mot.get_current_value() == 1 # not moved
|
|
assert set(task_queue.claims()) == {"mot"}
|
|
with task_queue.forced():
|
|
mot.set_target_value(0.5).wait()
|
|
assert mot.get_current_value() == 0.5
|
|
ns["gate"].set()
|
|
task.wait(5)
|
|
mot.set_target_value(2).wait() # the claim ended with the task
|
|
assert task_queue.claims() == {}
|
|
|
|
|
|
def test_the_tasks_own_moves_through_a_virtual_adjustable_are_not_blocked(queue, ns):
|
|
from eco.elements.adjustable import AdjustableVirtual, DummyAdjustable
|
|
|
|
a, b = DummyAdjustable(name="a"), DummyAdjustable(name="b")
|
|
both = AdjustableVirtual(
|
|
[a, b], lambda va, vb: va, lambda v: (v, 2 * v), name="both"
|
|
)
|
|
ns.update(both=both, gate=threading.Event())
|
|
task = queue.append("both.set_target_value(1).wait(); both.set_target_value(2).wait(); gate.wait(5)")
|
|
t0 = time.monotonic()
|
|
while b.get_current_value() != 4 and time.monotonic() - t0 < 5:
|
|
time.sleep(0.02)
|
|
assert (a.get_current_value(), b.get_current_value()) == (2, 4)
|
|
with pytest.raises(task_queue.MoveBlocked):
|
|
b.set_target_value(0).wait() # the motor underneath is claimed too
|
|
ns["gate"].set()
|
|
task.wait(5)
|
|
assert task.state == "done"
|
|
|
|
|
|
def test_a_queued_scan_claims_its_adjustables_from_its_start(scans_setup, monkeypatch):
|
|
scans, mot, _, counter = scans_setup
|
|
slow = threading.Event()
|
|
counter.acquire = lambda scan=None, Npulses=None, **kw: __import__(
|
|
"eco.acquisition.utilities", fromlist=["Acquisition"]
|
|
).Acquisition(acquire=lambda: slow.wait(5), hold=False)
|
|
q = TaskQueue("claim-scan", namespace={"__builtins__": __builtins__})
|
|
monkeypatch.setitem(task_queue.queues, "claim-scan", q)
|
|
try:
|
|
task = scans.ascan(mot, 0, 1, 2, 1, queue="claim-scan", return_at_end=False)
|
|
t0 = time.monotonic()
|
|
while task.scan is None and time.monotonic() - t0 < 10:
|
|
time.sleep(0.02)
|
|
with pytest.raises(task_queue.MoveBlocked):
|
|
mot.set_target_value(1.5).wait()
|
|
assert q.release(mot) is True
|
|
slow.set()
|
|
task.wait(20)
|
|
finally:
|
|
slow.set()
|
|
q.stop()
|
|
q.close()
|
|
|
|
|
|
# -- output: captured per task, mutable, recorded, shown live ----------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def session_out(monkeypatch):
|
|
"""sys.stdout as the session sees it, with the queue's router in front."""
|
|
import io
|
|
import sys
|
|
|
|
out = io.StringIO()
|
|
monkeypatch.setattr(sys, "stdout", out)
|
|
monkeypatch.setattr(sys, "stderr", out)
|
|
task_queue._install_output_router()
|
|
return out
|
|
|
|
|
|
def test_task_output_is_captured_including_threads_it_starts(queue, ns, session_out):
|
|
from eco.utilities.utilities import PropagatingThread
|
|
|
|
ns["PropagatingThread"] = PropagatingThread
|
|
task = queue.append(
|
|
"print('from the task')\n"
|
|
"t = PropagatingThread(target=print, args=('from its thread',)); t.start(); t.join()"
|
|
)
|
|
task.wait(5)
|
|
assert task.output == "from the task\nfrom its thread\n"
|
|
assert "from the task" in session_out.getvalue() # not muted: shown too
|
|
|
|
|
|
def test_muted_output_is_kept_but_failures_still_shown(queue, ns, session_out):
|
|
queue.mute()
|
|
task = queue.append("print('chatty')")
|
|
task.wait(5)
|
|
failing = queue.append("print('before'); 1 / 0")
|
|
with pytest.raises(ZeroDivisionError):
|
|
failing.wait(5)
|
|
shown = session_out.getvalue()
|
|
assert "chatty" not in shown and "before" not in shown
|
|
assert "failed: ZeroDivisionError" in shown
|
|
assert task.output == "chatty\n"
|
|
assert "failed: ZeroDivisionError" in failing.output
|
|
|
|
|
|
def test_record_writes_headers_and_output(queue, ns, session_out, tmp_path):
|
|
path = queue.record(tmp_path / "queue.log")
|
|
queue.append("print('step 1 of 2')").wait(5)
|
|
queue.stop_recording()
|
|
text = path.read_text()
|
|
assert "=== queue 'test', task #" in text and "started" in text
|
|
assert "step 1 of 2\n" in text
|
|
assert "done after" in text
|
|
|
|
|
|
def test_output_window_shows_earlier_and_new_output(queue, ns, session_out):
|
|
QtWidgets = pytest.importorskip("qtpy.QtWidgets")
|
|
from qtpy import QtCore
|
|
|
|
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
|
queue.append("print('earlier')").wait(5)
|
|
from eco.widgets.queue_output_qt import QueueOutputQt
|
|
|
|
window = QueueOutputQt(queue)
|
|
try:
|
|
queue.append("print('later')").wait(5)
|
|
loop = QtCore.QEventLoop()
|
|
QtCore.QTimer.singleShot(500, loop.quit)
|
|
loop.exec_()
|
|
text = window._text.toPlainText()
|
|
assert "earlier" in text and "later" in text
|
|
assert "queue 'test'" in window._status.text()
|
|
finally:
|
|
window.stop()
|
|
del app
|