mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 11:41:49 +02:00
refactor(clean-up): 1st generation widgets are removed
This commit is contained in:
@ -1,220 +0,0 @@
|
||||
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
|
||||
import os
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from bec_widgets.widgets import BECMonitor
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
|
||||
def load_test_config(config_name):
|
||||
"""Helper function to load config from yaml file."""
|
||||
config_path = os.path.join(os.path.dirname(__file__), "test_configs", f"{config_name}.yaml")
|
||||
with open(config_path, "r") as f:
|
||||
config = yaml.safe_load(f)
|
||||
return config
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def monitor(bec_dispatcher, qtbot, mocked_client):
|
||||
# client = MagicMock()
|
||||
widget = BECMonitor(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config_name, scan_type, number_of_plots",
|
||||
[
|
||||
("config_device", False, 2),
|
||||
("config_device_no_entry", False, 2),
|
||||
# ("config_scan", True, 4),
|
||||
],
|
||||
)
|
||||
def test_initialization_with_device_config(monitor, config_name, scan_type, number_of_plots):
|
||||
config = load_test_config(config_name)
|
||||
monitor.on_config_update(config)
|
||||
assert isinstance(monitor, BECMonitor)
|
||||
assert monitor.client is not None
|
||||
assert len(monitor.plot_data) == number_of_plots
|
||||
assert monitor.scan_types == scan_type
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config_initial,config_update",
|
||||
[("config_device", "config_scan"), ("config_scan", "config_device")],
|
||||
)
|
||||
def test_on_config_update(monitor, config_initial, config_update):
|
||||
config_initial = load_test_config(config_initial)
|
||||
config_update = load_test_config(config_update)
|
||||
# validated config has to be compared
|
||||
config_initial_validated = monitor.validator.validate_monitor_config(
|
||||
config_initial
|
||||
).model_dump()
|
||||
config_update_validated = monitor.validator.validate_monitor_config(config_update).model_dump()
|
||||
monitor.on_config_update(config_initial)
|
||||
assert monitor.config == config_initial_validated
|
||||
monitor.on_config_update(config_update)
|
||||
assert monitor.config == config_update_validated
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config_name, expected_num_columns, expected_plot_names, expected_coordinates",
|
||||
[
|
||||
("config_device", 1, ["BPM4i plots vs samx", "Gauss plots vs samx"], [(0, 0), (1, 0)]),
|
||||
(
|
||||
"config_scan",
|
||||
3,
|
||||
["Grid plot 1", "Grid plot 2", "Grid plot 3", "Grid plot 4"],
|
||||
[(0, 0), (0, 1), (0, 2), (1, 0)],
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_render_initial_plots(
|
||||
monitor, config_name, expected_num_columns, expected_plot_names, expected_coordinates
|
||||
):
|
||||
config = load_test_config(config_name)
|
||||
monitor.on_config_update(config)
|
||||
|
||||
# Validate number of columns
|
||||
assert monitor.plot_settings["num_columns"] == expected_num_columns
|
||||
|
||||
# Validate the plots are created correctly
|
||||
for expected_name in expected_plot_names:
|
||||
assert expected_name in monitor.plots.keys()
|
||||
|
||||
# Validate the grid_coordinates
|
||||
assert monitor.grid_coordinates == expected_coordinates
|
||||
|
||||
|
||||
def mock_getitem(dev_name):
|
||||
"""Helper function to mock the __getitem__ method of the 'dev'."""
|
||||
mock_instance = MagicMock()
|
||||
if dev_name == "samx":
|
||||
mock_instance._hints = "samx"
|
||||
elif dev_name == "bpm4i":
|
||||
mock_instance._hints = "bpm4i"
|
||||
elif dev_name == "gauss_bpm":
|
||||
mock_instance._hints = "gauss_bpm"
|
||||
|
||||
return mock_instance
|
||||
|
||||
|
||||
def mock_get_scan_storage(scan_id, data):
|
||||
"""Helper function to mock the __getitem__ method of the 'dev'."""
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_scan_storage.return_value = data
|
||||
return mock_instance
|
||||
|
||||
|
||||
# mocked messages and metadata
|
||||
msg_1 = {
|
||||
"data": {
|
||||
"samx": {"samx": {"value": 10}},
|
||||
"bpm4i": {"bpm4i": {"value": 5}},
|
||||
"gauss_bpm": {"gauss_bpm": {"value": 6}},
|
||||
"gauss_adc1": {"gauss_adc1": {"value": 8}},
|
||||
"gauss_adc2": {"gauss_adc2": {"value": 9}},
|
||||
},
|
||||
"scan_id": 1,
|
||||
}
|
||||
metadata_grid = {"scan_name": "grid_scan"}
|
||||
metadata_line = {"scan_name": "line_scan"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config_name, msg, metadata, expected_data",
|
||||
[
|
||||
# case: msg does not have 'scan_id'
|
||||
(
|
||||
"config_device",
|
||||
{"data": {}},
|
||||
{},
|
||||
{
|
||||
"scan_segment": {
|
||||
"bpm4i": {"bpm4i": []},
|
||||
"gauss_adc1": {"gauss_adc1": []},
|
||||
"gauss_adc2": {"gauss_adc2": []},
|
||||
"samx": {"samx": []},
|
||||
}
|
||||
},
|
||||
),
|
||||
# case: scan_types is false, msg contains all valid fields, and entry is present in config
|
||||
(
|
||||
"config_device",
|
||||
msg_1,
|
||||
{},
|
||||
{
|
||||
"scan_segment": {
|
||||
"bpm4i": {"bpm4i": [5]},
|
||||
"gauss_adc1": {"gauss_adc1": [8]},
|
||||
"gauss_adc2": {"gauss_adc2": [9]},
|
||||
"samx": {"samx": [10]},
|
||||
}
|
||||
},
|
||||
),
|
||||
# case: scan_types is false, msg contains all valid fields and entry is missing in config, should use hints
|
||||
(
|
||||
"config_device_no_entry",
|
||||
msg_1,
|
||||
{},
|
||||
{
|
||||
"scan_segment": {
|
||||
"bpm4i": {"bpm4i": [5]},
|
||||
"gauss_bpm": {"gauss_bpm": [6]},
|
||||
"samx": {"samx": [10]},
|
||||
}
|
||||
},
|
||||
),
|
||||
# case: scan_types is true, msg contains all valid fields, metadata contains scan "line_scan:"
|
||||
(
|
||||
"config_scan",
|
||||
msg_1,
|
||||
metadata_line,
|
||||
{
|
||||
"scan_segment": {
|
||||
"bpm4i": {"bpm4i": [5]},
|
||||
"gauss_adc1": {"gauss_adc1": [8]},
|
||||
"gauss_adc2": {"gauss_adc2": [9]},
|
||||
"gauss_bpm": {"gauss_bpm": [6]},
|
||||
"samx": {"samx": [10]},
|
||||
}
|
||||
},
|
||||
),
|
||||
(
|
||||
"config_scan",
|
||||
msg_1,
|
||||
metadata_grid,
|
||||
{
|
||||
"scan_segment": {
|
||||
"bpm4i": {"bpm4i": [5]},
|
||||
"gauss_adc1": {"gauss_adc1": [8]},
|
||||
"gauss_adc2": {"gauss_adc2": [9]},
|
||||
"gauss_bpm": {"gauss_bpm": [6]},
|
||||
"samx": {"samx": [10]},
|
||||
}
|
||||
},
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_on_scan_segment(monitor, config_name, msg, metadata, expected_data):
|
||||
config = load_test_config(config_name)
|
||||
monitor.on_config_update(config)
|
||||
|
||||
# Mock scan_storage.find_scan_by_ID
|
||||
mock_scan_data = MagicMock()
|
||||
mock_scan_data.data = {
|
||||
device_name: {
|
||||
entry: MagicMock(val=[msg["data"][device_name][entry]["value"]])
|
||||
for entry in msg["data"][device_name]
|
||||
}
|
||||
for device_name in msg["data"]
|
||||
}
|
||||
monitor.queue.scan_storage.find_scan_by_ID.return_value = mock_scan_data
|
||||
|
||||
monitor.on_scan_segment(msg, metadata)
|
||||
assert monitor.database == expected_data
|
||||
@ -1,178 +0,0 @@
|
||||
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
|
||||
import os
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from qtpy.QtWidgets import QTableWidgetItem, QTabWidget
|
||||
|
||||
from bec_widgets.widgets.monitor.config_dialog import ConfigDialog
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
|
||||
def load_test_config(config_name):
|
||||
"""Helper function to load config from yaml file."""
|
||||
config_path = os.path.join(os.path.dirname(__file__), "test_configs", f"{config_name}.yaml")
|
||||
with open(config_path, "r") as f:
|
||||
config = yaml.safe_load(f)
|
||||
return config
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def config_dialog(qtbot, mocked_client):
|
||||
client = mocked_client
|
||||
widget = ConfigDialog(client=client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
@pytest.mark.parametrize("config_name", ["config_device", "config_scan"])
|
||||
def test_load_config(config_dialog, config_name):
|
||||
config = load_test_config(config_name)
|
||||
config_dialog.load_config(config)
|
||||
|
||||
assert (
|
||||
config_dialog.comboBox_appearance.currentText()
|
||||
== config["plot_settings"]["background_color"]
|
||||
)
|
||||
assert config_dialog.spinBox_n_column.value() == config["plot_settings"]["num_columns"]
|
||||
assert config_dialog.comboBox_colormap.currentText() == config["plot_settings"]["colormap"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config_name, scan_mode",
|
||||
[("config_device", False), ("config_scan", True), ("config_device_no_entry", False)],
|
||||
)
|
||||
def test_initialization(config_dialog, config_name, scan_mode):
|
||||
config = load_test_config(config_name)
|
||||
config_dialog.load_config(config)
|
||||
|
||||
assert isinstance(config_dialog, ConfigDialog)
|
||||
assert (
|
||||
config_dialog.comboBox_appearance.currentText()
|
||||
== config["plot_settings"]["background_color"]
|
||||
)
|
||||
assert config_dialog.spinBox_n_column.value() == config["plot_settings"]["num_columns"]
|
||||
assert (config_dialog.comboBox_scanTypes.currentText() == "Enabled") == scan_mode
|
||||
assert (
|
||||
config_dialog.tabWidget_scan_types.count() > 0
|
||||
) # Ensures there's at least one tab created
|
||||
|
||||
# If there's a need to check the contents of the first tab (there has to be always at least one tab)
|
||||
first_tab = config_dialog.tabWidget_scan_types.widget(0)
|
||||
if scan_mode:
|
||||
assert (
|
||||
first_tab.findChild(QTabWidget, "tabWidget_plots") is not None
|
||||
) # Ensures plot tab widget exists in scan mode
|
||||
else:
|
||||
assert (
|
||||
first_tab.findChild(QTabWidget) is not None
|
||||
) # Ensures plot tab widget exists in default mode
|
||||
|
||||
|
||||
def test_edit_and_apply_config(config_dialog):
|
||||
config_device = load_test_config("config_device")
|
||||
config_dialog.load_config(config_device)
|
||||
|
||||
config_dialog.comboBox_appearance.setCurrentText("white")
|
||||
config_dialog.spinBox_n_column.setValue(2)
|
||||
config_dialog.comboBox_colormap.setCurrentText("viridis")
|
||||
|
||||
applied_config = config_dialog.apply_config()
|
||||
|
||||
assert applied_config["plot_settings"]["background_color"] == "white"
|
||||
assert applied_config["plot_settings"]["num_columns"] == 2
|
||||
assert applied_config["plot_settings"]["colormap"] == "viridis"
|
||||
|
||||
|
||||
def test_edit_and_apply_config_scan_mode(config_dialog):
|
||||
config_scan = load_test_config("config_scan")
|
||||
config_dialog.load_config(config_scan)
|
||||
|
||||
config_dialog.comboBox_appearance.setCurrentText("white")
|
||||
config_dialog.spinBox_n_column.setValue(2)
|
||||
config_dialog.comboBox_colormap.setCurrentText("viridis")
|
||||
config_dialog.comboBox_scanTypes.setCurrentText("Enabled")
|
||||
|
||||
applied_config = config_dialog.apply_config()
|
||||
|
||||
assert applied_config["plot_settings"]["background_color"] == "white"
|
||||
assert applied_config["plot_settings"]["num_columns"] == 2
|
||||
assert applied_config["plot_settings"]["colormap"] == "viridis"
|
||||
assert applied_config["plot_settings"]["scan_types"] is True
|
||||
|
||||
|
||||
def test_add_new_scan(config_dialog):
|
||||
# Ensure the tab count is initially 1 (from the default config)
|
||||
assert config_dialog.tabWidget_scan_types.count() == 1
|
||||
|
||||
# Add a new scan tab
|
||||
config_dialog.add_new_scan_tab(config_dialog.tabWidget_scan_types, "Test Scan Tab")
|
||||
|
||||
# Ensure the tab count is now 2
|
||||
assert config_dialog.tabWidget_scan_types.count() == 2
|
||||
|
||||
# Ensure the new tab has the correct name
|
||||
assert config_dialog.tabWidget_scan_types.tabText(1) == "Test Scan Tab"
|
||||
|
||||
|
||||
def test_add_new_plot_and_modify(config_dialog):
|
||||
# Ensure the tab count is initially 1 and it is called "Default"
|
||||
assert config_dialog.tabWidget_scan_types.count() == 1
|
||||
assert config_dialog.tabWidget_scan_types.tabText(0) == "Default"
|
||||
|
||||
# Get the first tab (which should be a scan tab)
|
||||
scan_tab = config_dialog.tabWidget_scan_types.widget(0)
|
||||
|
||||
# Ensure the plot tab count is initially 1 and it is called "Plot 1"
|
||||
tabWidget_plots = scan_tab.findChild(QTabWidget)
|
||||
assert tabWidget_plots.count() == 1
|
||||
assert tabWidget_plots.tabText(0) == "Plot 1"
|
||||
|
||||
# Add a new plot tab
|
||||
config_dialog.add_new_plot_tab(scan_tab)
|
||||
|
||||
# Ensure the plot tab count is now 2
|
||||
assert tabWidget_plots.count() == 2
|
||||
|
||||
# Ensure the new tab has the correct name
|
||||
assert tabWidget_plots.tabText(1) == "Plot 2"
|
||||
|
||||
# Access the new plot tab
|
||||
new_plot_tab = tabWidget_plots.widget(1)
|
||||
|
||||
# Modify the line edits within the new plot tab
|
||||
new_plot_tab.ui.lineEdit_plot_title.setText("Modified Plot Title")
|
||||
new_plot_tab.ui.lineEdit_x_label.setText("Modified X Label")
|
||||
new_plot_tab.ui.lineEdit_y_label.setText("Modified Y Label")
|
||||
new_plot_tab.ui.lineEdit_x_name.setText("Modified X Name")
|
||||
new_plot_tab.ui.lineEdit_x_entry.setText("Modified X Entry")
|
||||
|
||||
# Modify the table for signals
|
||||
config_dialog.add_new_signal(new_plot_tab.ui.tableWidget_y_signals)
|
||||
|
||||
table = new_plot_tab.ui.tableWidget_y_signals
|
||||
assert table.rowCount() == 1 # Ensure the new row is added
|
||||
|
||||
row_position = table.rowCount() - 1
|
||||
|
||||
# Modify the first row
|
||||
table.setItem(row_position, 0, QTableWidgetItem("New Signal Name"))
|
||||
table.setItem(row_position, 1, QTableWidgetItem("New Signal Entry"))
|
||||
|
||||
# Apply the configuration
|
||||
config = config_dialog.apply_config()
|
||||
|
||||
# Check if the modifications are reflected in the configuration
|
||||
modified_plot_config = config["plot_data"][1] # Access the second plot in the plot_data list
|
||||
sources = modified_plot_config["sources"][0] # Access the first source in the sources list
|
||||
|
||||
assert modified_plot_config["plot_name"] == "Modified Plot Title"
|
||||
assert modified_plot_config["x_label"] == "Modified X Label"
|
||||
assert modified_plot_config["y_label"] == "Modified Y Label"
|
||||
assert sources["signals"]["x"][0]["name"] == "Modified X Name"
|
||||
assert sources["signals"]["x"][0]["entry"] == "Modified X Entry"
|
||||
assert sources["signals"]["y"][0]["name"] == "New Signal Name"
|
||||
assert sources["signals"]["y"][0]["entry"] == "New Signal Entry"
|
||||
@ -1,171 +0,0 @@
|
||||
# pylint: disable = no-name-in-module,missing-module-docstring, missing-function-docstring
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from bec_widgets.widgets import MotorMap
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
CONFIG_DEFAULT = {
|
||||
"plot_settings": {
|
||||
"colormap": "Greys",
|
||||
"scatter_size": 5,
|
||||
"max_points": 1000,
|
||||
"num_dim_points": 100,
|
||||
"precision": 2,
|
||||
"num_columns": 1,
|
||||
"background_value": 25,
|
||||
},
|
||||
"motors": [
|
||||
{
|
||||
"plot_name": "Motor Map",
|
||||
"x_label": "Motor X",
|
||||
"y_label": "Motor Y",
|
||||
"signals": {
|
||||
"x": [{"name": "samx", "entry": "samx"}],
|
||||
"y": [{"name": "samy", "entry": "samy"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"plot_name": "Motor Map 2 ",
|
||||
"x_label": "Motor X",
|
||||
"y_label": "Motor Y",
|
||||
"signals": {
|
||||
"x": [{"name": "aptrx", "entry": "aptrx"}],
|
||||
"y": [{"name": "aptry", "entry": "aptry"}],
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
CONFIG_ONE_DEVICE = {
|
||||
"plot_settings": {
|
||||
"colormap": "Greys",
|
||||
"scatter_size": 5,
|
||||
"max_points": 1000,
|
||||
"num_dim_points": 100,
|
||||
"precision": 2,
|
||||
"num_columns": 1,
|
||||
"background_value": 25,
|
||||
},
|
||||
"motors": [
|
||||
{
|
||||
"plot_name": "Motor Map",
|
||||
"x_label": "Motor X",
|
||||
"y_label": "Motor Y",
|
||||
"signals": {
|
||||
"x": [{"name": "samx", "entry": "samx"}],
|
||||
"y": [{"name": "samy", "entry": "samy"}],
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def motor_map(qtbot, mocked_client):
|
||||
widget = MotorMap(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
def test_motor_limits_initialization(motor_map):
|
||||
# Example test to check if motor limits are correctly initialized
|
||||
expected_limits = {"samx": [-10, 10], "samy": [-5, 5]}
|
||||
for motor_name, expected_limit in expected_limits.items():
|
||||
actual_limit = motor_map._get_motor_limit(motor_name)
|
||||
assert actual_limit == expected_limit
|
||||
|
||||
|
||||
def test_motor_initial_position(motor_map):
|
||||
motor_map.precision = 2
|
||||
|
||||
motor_map_dev = motor_map.client.device_manager.devices
|
||||
|
||||
# Example test to check if motor initial positions are correctly initialized
|
||||
expected_positions = {
|
||||
("samx", "samx"): motor_map_dev["samx"].read()["samx"]["value"],
|
||||
("samy", "samy"): motor_map_dev["samy"].read()["samy"]["value"],
|
||||
("aptrx", "aptrx"): motor_map_dev["aptrx"].read()["aptrx"]["value"],
|
||||
("aptry", "aptry"): motor_map_dev["aptry"].read()["aptry"]["value"],
|
||||
}
|
||||
for (motor_name, entry), expected_position in expected_positions.items():
|
||||
actual_position = motor_map._get_motor_init_position(motor_name, entry)
|
||||
assert actual_position == expected_position
|
||||
|
||||
|
||||
@pytest.mark.parametrize("config, number_of_plots", [(CONFIG_DEFAULT, 2), (CONFIG_ONE_DEVICE, 1)])
|
||||
def test_initialization(motor_map, config, number_of_plots):
|
||||
config_load = config
|
||||
motor_map.on_config_update(config_load)
|
||||
assert isinstance(motor_map, MotorMap)
|
||||
assert motor_map.client is not None
|
||||
assert motor_map.config == config_load
|
||||
assert len(motor_map.plot_data) == number_of_plots
|
||||
|
||||
|
||||
def test_motor_movement_updates_position_and_database(motor_map):
|
||||
motor_map.on_config_update(CONFIG_DEFAULT)
|
||||
|
||||
# Initial positions
|
||||
initial_position_samx = 2.0
|
||||
initial_position_samy = 3.0
|
||||
|
||||
# Set initial positions in the mocked database
|
||||
motor_map.database["samx"]["samx"] = [initial_position_samx]
|
||||
motor_map.database["samy"]["samy"] = [initial_position_samy]
|
||||
|
||||
# Simulate motor movement for 'samx' only
|
||||
new_position_samx = 4.0
|
||||
motor_map.on_device_readback({"signals": {"samx": {"value": new_position_samx}}})
|
||||
|
||||
# Verify database update for 'samx'
|
||||
assert motor_map.database["samx"]["samx"] == [initial_position_samx, new_position_samx]
|
||||
|
||||
# Verify 'samy' retains its last known position
|
||||
assert motor_map.database["samy"]["samy"] == [initial_position_samy, initial_position_samy]
|
||||
|
||||
|
||||
def test_scatter_plot_rendering(motor_map):
|
||||
motor_map.on_config_update(CONFIG_DEFAULT)
|
||||
# Set initial positions
|
||||
initial_position_samx = 2.0
|
||||
initial_position_samy = 3.0
|
||||
motor_map.database["samx"]["samx"] = [initial_position_samx]
|
||||
motor_map.database["samy"]["samy"] = [initial_position_samy]
|
||||
|
||||
# Simulate motor movement for 'samx' only
|
||||
new_position_samx = 4.0
|
||||
motor_map.on_device_readback({"signals": {"samx": {"value": new_position_samx}}})
|
||||
motor_map._update_plots()
|
||||
|
||||
# Get the scatter plot item
|
||||
plot_name = "Motor Map" # Update as per your actual plot name
|
||||
scatter_plot_item = motor_map.curves_data[plot_name]["pos"]
|
||||
|
||||
# Check the scatter plot item properties
|
||||
assert len(scatter_plot_item.data) > 0, "Scatter plot data is empty"
|
||||
x_data = scatter_plot_item.data["x"]
|
||||
y_data = scatter_plot_item.data["y"]
|
||||
assert x_data[-1] == new_position_samx, "Scatter plot X data not updated correctly"
|
||||
assert (
|
||||
y_data[-1] == initial_position_samy
|
||||
), "Scatter plot Y data should retain last known position"
|
||||
|
||||
|
||||
def test_plot_visualization_consistency(motor_map):
|
||||
motor_map.on_config_update(CONFIG_DEFAULT)
|
||||
# Simulate updating the plot with new data
|
||||
motor_map.on_device_readback({"signals": {"samx": {"value": 5}}})
|
||||
motor_map.on_device_readback({"signals": {"samy": {"value": 9}}})
|
||||
motor_map._update_plots()
|
||||
|
||||
plot_name = "Motor Map"
|
||||
scatter_plot_item = motor_map.curves_data[plot_name]["pos"]
|
||||
|
||||
# Check if the scatter plot reflects the new data correctly
|
||||
assert (
|
||||
scatter_plot_item.data["x"][-1] == 5 and scatter_plot_item.data["y"][-1] == 9
|
||||
), "Plot not updated correctly with new data"
|
||||
@ -1,110 +0,0 @@
|
||||
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from bec_widgets.validation.monitor_config_validator import (
|
||||
AxisSignal,
|
||||
MonitorConfigValidator,
|
||||
PlotConfig,
|
||||
Signal,
|
||||
)
|
||||
|
||||
from .test_bec_monitor import mocked_client
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def setup_devices(mocked_client):
|
||||
MonitorConfigValidator.devices = mocked_client.device_manager.devices
|
||||
|
||||
|
||||
def test_signal_validation_name_missing(setup_devices):
|
||||
with pytest.raises(ValidationError) as excinfo:
|
||||
Signal(name=None)
|
||||
errors = excinfo.value.errors()
|
||||
assert len(errors) == 1
|
||||
assert errors[0]["type"] == "no_device_name"
|
||||
assert "Device name must be provided" in str(excinfo.value)
|
||||
|
||||
|
||||
def test_signal_validation_name_not_in_bec(setup_devices):
|
||||
with pytest.raises(ValidationError) as excinfo:
|
||||
Signal(name="non_existent_device")
|
||||
errors = excinfo.value.errors()
|
||||
assert len(errors) == 1
|
||||
assert errors[0]["type"] == "no_device_bec"
|
||||
assert 'Device "non_existent_device" not found in current BEC session' in str(excinfo.value)
|
||||
|
||||
|
||||
def test_signal_validation_entry_not_in_device(setup_devices):
|
||||
with pytest.raises(ValidationError) as excinfo:
|
||||
Signal(name="samx", entry="non_existent_entry")
|
||||
|
||||
errors = excinfo.value.errors()
|
||||
assert len(errors) == 1
|
||||
assert errors[0]["type"] == "no_entry_for_device"
|
||||
assert 'Entry "non_existent_entry" not found in device "samx" signals' in errors[0]["msg"]
|
||||
|
||||
|
||||
def test_signal_validation_success(setup_devices):
|
||||
signal = Signal(name="samx")
|
||||
assert signal.name == "samx"
|
||||
|
||||
|
||||
def test_plot_config_x_axis_signal_validation(setup_devices):
|
||||
# Setup a valid signal
|
||||
valid_signal = Signal(name="samx")
|
||||
|
||||
with pytest.raises(ValidationError) as excinfo:
|
||||
AxisSignal(x=[valid_signal, valid_signal], y=[valid_signal, valid_signal])
|
||||
|
||||
errors = excinfo.value.errors()
|
||||
assert len(errors) == 1
|
||||
assert errors[0]["type"] == "x_axis_multiple_signals"
|
||||
assert "There must be exactly one signal for x axis" in errors[0]["msg"]
|
||||
|
||||
|
||||
def test_plot_config_unsupported_source_type(setup_devices):
|
||||
with pytest.raises(ValidationError) as excinfo:
|
||||
PlotConfig(sources=[{"type": "unsupported_type", "signals": {}}])
|
||||
|
||||
errors = excinfo.value.errors()
|
||||
print(errors)
|
||||
assert len(errors) == 1
|
||||
assert errors[0]["type"] == "literal_error"
|
||||
|
||||
|
||||
def test_plot_config_no_source_type_provided(setup_devices):
|
||||
with pytest.raises(ValidationError) as excinfo:
|
||||
PlotConfig(sources=[{"signals": {}}])
|
||||
|
||||
errors = excinfo.value.errors()
|
||||
assert len(errors) == 1
|
||||
assert errors[0]["type"] == "missing"
|
||||
|
||||
|
||||
def test_plot_config_history_source_type(setup_devices):
|
||||
history_source = {
|
||||
"type": "history",
|
||||
"scan_id": "valid_scan_id",
|
||||
"signals": {"x": [{"name": "samx"}], "y": [{"name": "samx"}]},
|
||||
}
|
||||
|
||||
plot_config = PlotConfig(sources=[history_source])
|
||||
|
||||
assert len(plot_config.sources) == 1
|
||||
assert plot_config.sources[0].type == "history"
|
||||
assert plot_config.sources[0].scan_id == "valid_scan_id"
|
||||
|
||||
|
||||
def test_plot_config_redis_source_type(setup_devices):
|
||||
history_source = {
|
||||
"type": "redis",
|
||||
"endpoint": "valid_endpoint",
|
||||
"update": "append",
|
||||
"signals": {"x": [{"name": "samx"}], "y": [{"name": "samx"}]},
|
||||
}
|
||||
|
||||
plot_config = PlotConfig(sources=[history_source])
|
||||
|
||||
assert len(plot_config.sources) == 1
|
||||
assert plot_config.sources[0].type == "redis"
|
||||
Reference in New Issue
Block a user