mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-06-05 21:08:40 +02:00
315 lines
11 KiB
Python
315 lines
11 KiB
Python
from unittest import mock
|
||
|
||
import numpy as np
|
||
import pytest
|
||
from bec_lib.endpoints import MessageEndpoints
|
||
|
||
from bec_widgets.utils.bec_widget import BECWidget
|
||
from bec_widgets.widgets.progress.bec_progressbar.bec_progressbar import (
|
||
BECProgressBar,
|
||
ProgressState,
|
||
)
|
||
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar import (
|
||
ProgressSource,
|
||
ProgressTask,
|
||
ScanProgressBar,
|
||
)
|
||
|
||
from .client_mocks import mocked_client
|
||
|
||
|
||
@pytest.fixture
|
||
def scan_progressbar(qtbot, mocked_client):
|
||
widget = ScanProgressBar(client=mocked_client)
|
||
qtbot.addWidget(widget)
|
||
qtbot.waitExposed(widget)
|
||
yield widget
|
||
|
||
|
||
def test_progress_task_basic():
|
||
"""percentage, remaining, and formatted time helpers behave as expected."""
|
||
task = ProgressTask(parent=None, value=50, max_value=100, done=False)
|
||
task.timer.stop() # we don’t want the timer ticking in tests
|
||
task._elapsed_time = 10 # simulate 10 s elapsed
|
||
|
||
# 50 / 100 ⇒ 50 %
|
||
assert task.percentage == 50
|
||
|
||
# speed = value / elapsed = 5 steps / s
|
||
assert np.isclose(task.speed, 5)
|
||
|
||
# remaining steps = 50 ; time_remaining ≈ 10 s
|
||
assert task.remaining == 50
|
||
assert task.time_remaining == "00:00:10"
|
||
|
||
# time_elapsed formatting
|
||
assert task.time_elapsed == "00:00:10"
|
||
|
||
|
||
def test_progress_task_elapsed_time_uses_monotonic_clock(monkeypatch):
|
||
times = iter([100.0, 102.5])
|
||
monkeypatch.setattr(
|
||
"bec_widgets.widgets.progress.progress_backend.time.monotonic", lambda: next(times)
|
||
)
|
||
task = ProgressTask(parent=None)
|
||
task.timer.stop()
|
||
|
||
task.update_elapsed_time()
|
||
|
||
assert task._elapsed_time == 2.5
|
||
assert task.time_elapsed == "00:00:02"
|
||
|
||
|
||
def test_scan_progressbar_initialization(scan_progressbar):
|
||
assert isinstance(scan_progressbar, ScanProgressBar)
|
||
assert isinstance(scan_progressbar.progressbar, BECProgressBar)
|
||
|
||
|
||
def test_scan_progressbar_passes_dynamic_stylesheet_setting(qtbot, mocked_client):
|
||
widget = ScanProgressBar(client=mocked_client, enable_dynamic_stylesheet=False)
|
||
qtbot.addWidget(widget)
|
||
|
||
assert widget.progressbar.enable_dynamic_stylesheet is False
|
||
|
||
|
||
def test_scan_progressbar_starts_from_scan_progress_before_queue_update(scan_progressbar):
|
||
scan_progressbar._clear_task(emit_finished=False)
|
||
|
||
scan_progressbar.on_progress_update(
|
||
{"value": 3, "max_value": 10, "done": False}, metadata={"RID": "live-rid"}
|
||
)
|
||
|
||
assert scan_progressbar.task is not None
|
||
assert scan_progressbar._active_scan_id == "live-rid"
|
||
assert scan_progressbar.progressbar._user_value == 3
|
||
assert scan_progressbar.progressbar._user_maximum == 10
|
||
|
||
|
||
def test_update_labels_content(scan_progressbar):
|
||
"""update_labels() reflects ProgressTask time strings on the UI."""
|
||
# fabricate a task with known timings
|
||
task = ProgressTask(parent=scan_progressbar, value=30, max_value=100, done=False)
|
||
task.timer.stop()
|
||
task._elapsed_time = 50
|
||
scan_progressbar.task = task
|
||
|
||
scan_progressbar.update_labels()
|
||
|
||
assert scan_progressbar.ui.elapsed_time_label.text() == "00:00:50"
|
||
assert scan_progressbar.ui.remaining_time_label.text() == "00:01:57"
|
||
|
||
|
||
def test_on_progress_update(qtbot, scan_progressbar):
|
||
"""
|
||
on_progress_update() should forward new values to the embedded
|
||
BECProgressBar and keep ProgressTask in sync.
|
||
"""
|
||
task = ProgressTask(parent=scan_progressbar, value=0, max_value=100, done=False)
|
||
task.timer.stop()
|
||
scan_progressbar.task = task
|
||
|
||
msg = {"value": 20, "max_value": 100, "done": False}
|
||
scan_progressbar.on_progress_update(msg, metadata={"status": "open"})
|
||
|
||
qtbot.wait(200)
|
||
bar = scan_progressbar.progressbar
|
||
assert bar._user_value == 20
|
||
assert bar._user_maximum == 100
|
||
# state reflects BEC status
|
||
assert bar.state is ProgressState.NORMAL
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"status, value, max_val, expected_state",
|
||
[
|
||
("open", 10, 100, ProgressState.NORMAL),
|
||
("paused", 25, 100, ProgressState.PAUSED),
|
||
("aborted", 30, 100, ProgressState.INTERRUPTED),
|
||
("halt", 40, 100, ProgressState.PAUSED),
|
||
("halted", 40, 100, ProgressState.PAUSED),
|
||
("closed", 100, 100, ProgressState.COMPLETED),
|
||
("user_completed", 40, 100, ProgressState.PAUSED),
|
||
],
|
||
)
|
||
def test_state_mapping_during_updates(
|
||
qtbot, scan_progressbar, status, value, max_val, expected_state
|
||
):
|
||
"""ScanProgressBar should translate BEC status → ProgressState consistently."""
|
||
task = ProgressTask(parent=scan_progressbar, value=0, max_value=max_val, done=False)
|
||
task.timer.stop()
|
||
scan_progressbar.task = task
|
||
|
||
scan_progressbar.on_progress_update(
|
||
{"value": value, "max_value": max_val, "done": status == "closed"},
|
||
metadata={"status": status},
|
||
)
|
||
|
||
assert scan_progressbar.progressbar.state is expected_state
|
||
|
||
|
||
def test_aborted_done_scan_keeps_partial_progress(scan_progressbar):
|
||
scan_progressbar.on_progress_update(
|
||
{"value": 4, "max_value": 10, "done": True},
|
||
metadata={"scan_id": "scan-1", "RID": "rid-1", "status": "aborted"},
|
||
)
|
||
|
||
assert scan_progressbar.progressbar._user_value == 4
|
||
assert scan_progressbar.progressbar._user_maximum == 10
|
||
assert scan_progressbar.progressbar.state is ProgressState.INTERRUPTED
|
||
assert scan_progressbar.task is None
|
||
|
||
|
||
def test_source_label_updates(scan_progressbar):
|
||
"""update_source_label() renders correct text for both progress sources."""
|
||
# device progress
|
||
scan_progressbar.update_source_label(ProgressSource.DEVICE_PROGRESS, device="motor")
|
||
assert scan_progressbar.ui.source_label.text() == "Device motor"
|
||
|
||
# scan progress (needs a scan_number for deterministic text)
|
||
scan_progressbar.scan_number = 5
|
||
scan_progressbar.update_source_label(ProgressSource.SCAN_PROGRESS)
|
||
assert scan_progressbar.ui.source_label.text() == "Scan 5"
|
||
|
||
|
||
def test_source_label_update_logs_only_on_text_change(scan_progressbar):
|
||
scan_progressbar.scan_number = 5
|
||
|
||
with mock.patch(
|
||
"bec_widgets.widgets.progress.scan_progressbar.scan_progressbar.logger.info"
|
||
) as mock_info:
|
||
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
|
||
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
|
||
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
|
||
|
||
mock_info.assert_called_once_with("Set progress source to Scan 5")
|
||
|
||
|
||
def test_set_progress_source_connections(scan_progressbar, monkeypatch):
|
||
""" """
|
||
|
||
connect_calls = []
|
||
disconnect_calls = []
|
||
|
||
def fake_connect(slot, endpoint):
|
||
connect_calls.append(endpoint)
|
||
|
||
def fake_disconnect(slot, endpoint):
|
||
disconnect_calls.append(endpoint)
|
||
|
||
# Patch dispatcher methods
|
||
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", fake_connect)
|
||
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "disconnect_slot", fake_disconnect)
|
||
|
||
# switch to SCAN_PROGRESS
|
||
scan_progressbar.scan_number = 7
|
||
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
|
||
|
||
assert scan_progressbar._progress_source == ProgressSource.SCAN_PROGRESS
|
||
assert scan_progressbar.ui.source_label.text() == "Scan 7"
|
||
assert connect_calls == []
|
||
assert disconnect_calls == []
|
||
|
||
# switch to DEVICE_PROGRESS
|
||
device = "motor"
|
||
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device=device)
|
||
|
||
assert scan_progressbar._progress_source == ProgressSource.DEVICE_PROGRESS
|
||
assert scan_progressbar.ui.source_label.text() == f"Device {device}"
|
||
assert connect_calls[-1] == MessageEndpoints.device_progress(device=device)
|
||
assert disconnect_calls == [MessageEndpoints.scan_progress()]
|
||
|
||
# calling again with the SAME source should not add new connect calls
|
||
prev_connect_count = len(connect_calls)
|
||
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device=device)
|
||
assert len(connect_calls) == prev_connect_count, "No extra connect made for same source"
|
||
|
||
|
||
def test_set_progress_source_disconnects_previous_device_subscription(
|
||
scan_progressbar, monkeypatch
|
||
):
|
||
|
||
disconnect_calls = []
|
||
|
||
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", lambda *args: None)
|
||
monkeypatch.setattr(
|
||
scan_progressbar.bec_dispatcher,
|
||
"disconnect_slot",
|
||
lambda slot, endpoint: disconnect_calls.append(endpoint),
|
||
)
|
||
|
||
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
|
||
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor2")
|
||
|
||
assert disconnect_calls == [
|
||
MessageEndpoints.scan_progress(),
|
||
MessageEndpoints.device_progress(device="motor1"),
|
||
]
|
||
|
||
|
||
def test_set_progress_source_disconnects_device_when_switching_to_scan(
|
||
scan_progressbar, monkeypatch
|
||
):
|
||
|
||
disconnect_calls = []
|
||
|
||
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", lambda *args: None)
|
||
monkeypatch.setattr(
|
||
scan_progressbar.bec_dispatcher,
|
||
"disconnect_slot",
|
||
lambda slot, endpoint: disconnect_calls.append(endpoint),
|
||
)
|
||
|
||
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
|
||
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
|
||
|
||
assert disconnect_calls == [
|
||
MessageEndpoints.scan_progress(),
|
||
MessageEndpoints.device_progress(device="motor1"),
|
||
]
|
||
|
||
|
||
def test_cleanup_disconnects_active_device_subscription(scan_progressbar, monkeypatch):
|
||
|
||
disconnect_calls = []
|
||
|
||
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", lambda *args: None)
|
||
monkeypatch.setattr(
|
||
scan_progressbar.bec_dispatcher,
|
||
"disconnect_slot",
|
||
lambda slot, endpoint: disconnect_calls.append(endpoint),
|
||
)
|
||
monkeypatch.setattr(scan_progressbar.progressbar, "close", lambda: None)
|
||
monkeypatch.setattr(scan_progressbar.progressbar, "deleteLater", lambda: None)
|
||
monkeypatch.setattr(BECWidget, "cleanup", lambda self: None)
|
||
|
||
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
|
||
with (
|
||
mock.patch.object(scan_progressbar, "close", wraps=scan_progressbar.close) as close_mock,
|
||
mock.patch.object(
|
||
scan_progressbar, "deleteLater", wraps=scan_progressbar.deleteLater
|
||
) as delete_later_mock,
|
||
):
|
||
ScanProgressBar.cleanup(scan_progressbar)
|
||
|
||
assert disconnect_calls == [
|
||
MessageEndpoints.scan_progress(),
|
||
MessageEndpoints.device_progress(device="motor1"),
|
||
]
|
||
assert scan_progressbar._progress_source is None
|
||
assert scan_progressbar._progress_device is None
|
||
close_mock.assert_not_called()
|
||
delete_later_mock.assert_not_called()
|
||
|
||
|
||
def test_cleanup_stops_active_task(scan_progressbar, monkeypatch):
|
||
monkeypatch.setattr(BECWidget, "cleanup", lambda self: None)
|
||
scan_progressbar.task = ProgressTask(parent=scan_progressbar)
|
||
scan_progressbar._active_scan_id = "scan-1"
|
||
timer = scan_progressbar.task.timer
|
||
|
||
ScanProgressBar.cleanup(scan_progressbar)
|
||
|
||
assert not timer.isActive()
|
||
assert scan_progressbar.task is None
|
||
assert scan_progressbar._active_scan_id is None
|